CLOUDSTACK-8822 - Replacing Runnable by Callable in the Taks and NioConnection classes

   - All the sub-classes were also updated according to the changes in the super-classes
   - There were also code formatting changes

Author: wilderrodrigues

agent/src/com/cloud/agent/Agent.java

@@ -103,6 +103,8 @@
 import com.cloud.utils.db.TransactionLegacy;
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.utils.exception.HypervisorVersionChangedException;
+import com.cloud.utils.exception.NioConnectionException;
+import com.cloud.utils.exception.TaskExecutionException;
 import com.cloud.utils.fsm.NoTransitionException;
 import com.cloud.utils.fsm.StateMachine2;
 import com.cloud.utils.nio.HandlerFactory;
@@ -593,7 +595,11 @@ public boolean start() {
         startDirectlyConnectedHosts();
 
         if (_connection != null) {
-            _connection.start();
+            try {
+                _connection.start();
+            } catch (final NioConnectionException e) {
+                s_logger.error("Error when connecting to the NioServer!", e);
+            }
         }
 
         _monitorExecutor.scheduleWithFixedDelay(new MonitorTask(), PingInterval.value(), PingInterval.value(), TimeUnit.SECONDS);
@@ -827,7 +833,7 @@ protected boolean handleDisconnectWithInvestigation(final AgentAttache attache,
                 Status determinedState = investigate(attache);
                 // if state cannot be determined do nothing and bail out
                 if (determinedState == null) {
-                    if (((System.currentTimeMillis() >> 10) - host.getLastPinged()) > AlertWait.value()) {
+                    if ((System.currentTimeMillis() >> 10) - host.getLastPinged() > AlertWait.value()) {
                         s_logger.warn("Agent " + hostId + " state cannot be determined for more than " + AlertWait + "(" + AlertWait.value() + ") seconds, will go to Alert state");
                         determinedState = Status.Alert;
                     } else {
@@ -840,7 +846,7 @@ protected boolean handleDisconnectWithInvestigation(final AgentAttache attache,
                 s_logger.info("The agent " + hostId + " state determined is " + determinedState);
 
                 if (determinedState == Status.Down) {
-                    String message = "Host is down: " + host.getId() + "-" + host.getName() + ". Starting HA on the VMs";
+                    final String message = "Host is down: " + host.getId() + "-" + host.getName() + ". Starting HA on the VMs";
                     s_logger.error(message);
                     if (host.getType() != Host.Type.SecondaryStorage && host.getType() != Host.Type.ConsoleProxy) {
                         _alertMgr.sendAlert(AlertManager.AlertType.ALERT_TYPE_HOST, host.getDataCenterId(), host.getPodId(), "Host down, " + host.getId(), message);
@@ -1299,7 +1305,7 @@ protected void processResponse(final Link link, final Response response) {
         }
 
         @Override
-        protected void doTask(final Task task) throws Exception {
+        protected void doTask(final Task task) throws TaskExecutionException {
             final TransactionLegacy txn = TransactionLegacy.open(TransactionLegacy.CLOUD_DB);
             try {
                 final Type type = task.getType();
@@ -1315,6 +1321,10 @@ protected void doTask(final Task task) throws Exception {
                     } catch (final UnsupportedVersionException e) {
                         s_logger.warn(e.getMessage());
                         // upgradeAgent(task.getLink(), data, e.getReason());
+                    } catch (final ClassNotFoundException e) {
+                        final String message = String.format("Exception occured when executing taks! Error '%s'", e.getMessage());
+                        s_logger.error(message);
+                        throw new TaskExecutionException(message, e);
                     }
                 } else if (type == Task.Type.CONNECT) {
                 } else if (type == Task.Type.DISCONNECT) {

engine/orchestration/src/com/cloud/agent/manager/AgentManagerImpl.java

@@ -66,4 +66,6 @@ public interface SerialVersionUID {
     public static final long UnableDeleteHostException = Base | 0x29;
     public static final long AffinityConflictException = Base | 0x2a;
     public static final long JobCancellationException = Base | 0x2b;
+    public static final long NioConnectionException = Base | 0x2c;
+    public static final long TaskExecutionException = Base | 0x2d;
 }

engine/orchestration/src/com/cloud/agent/manager/ClusteredAgentManagerImpl.java

@@ -0,0 +1,48 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+package com.cloud.utils.exception;
+
+import com.cloud.utils.SerialVersionUID;
+
+/**
+ * Used by the NioConnection class to wrap-up its exceptions.
+ */
+public class NioConnectionException extends Exception {
+    private static final long serialVersionUID = SerialVersionUID.NioConnectionException;
+
+    protected int csErrorCode;
+
+    public NioConnectionException(final String msg, final Throwable cause) {
+        super(msg, cause);
+        setCSErrorCode(CSExceptionErrorCode.getCSErrCode(this.getClass().getName()));
+    }
+
+    public NioConnectionException(final String msg) {
+        super(msg);
+    }
+
+    public void setCSErrorCode(final int cserrcode) {
+        csErrorCode = cserrcode;
+    }
+
+    public int getCSErrorCode() {
+        return csErrorCode;
+    }
+}

utils/src/main/java/com/cloud/utils/SerialVersionUID.java

@@ -0,0 +1,48 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+package com.cloud.utils.exception;
+
+import com.cloud.utils.SerialVersionUID;
+
+/**
+ * Used by the Task class to wrap-up its exceptions.
+ */
+public class TaskExecutionException extends Exception {
+    private static final long serialVersionUID = SerialVersionUID.NioConnectionException;
+
+    protected int csErrorCode;
+
+    public TaskExecutionException(final String msg, final Throwable cause) {
+        super(msg, cause);
+        setCSErrorCode(CSExceptionErrorCode.getCSErrCode(this.getClass().getName()));
+    }
+
+    public TaskExecutionException(final String msg) {
+        super(msg);
+    }
+
+    public void setCSErrorCode(final int cserrcode) {
+        csErrorCode = cserrcode;
+    }
+
+    public int getCSErrorCode() {
+        return csErrorCode;
+    }
+}

utils/src/main/java/com/cloud/utils/exception/NioConnectionException.java

@@ -29,9 +29,8 @@
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
 
-import org.apache.log4j.Logger;
-
 import org.apache.cloudstack.utils.security.SSLUtils;
+import org.apache.log4j.Logger;
 
 public class NioClient extends NioConnection {
     private static final Logger s_logger = Logger.getLogger(NioClient.class);
@@ -40,12 +39,12 @@ public class NioClient extends NioConnection {
     protected String _bindAddress;
     protected SocketChannel _clientConnection;
 
-    public NioClient(String name, String host, int port, int workers, HandlerFactory factory) {
+    public NioClient(final String name, final String host, final int port, final int workers, final HandlerFactory factory) {
         super(name, port, workers, factory);
         _host = host;
     }
 
-    public void setBindAddress(String ipAddress) {
+    public void setBindAddress(final String ipAddress) {
         _bindAddress = ipAddress;
     }
 
@@ -62,18 +61,18 @@ protected void init() throws IOException {
             if (_bindAddress != null) {
                 s_logger.info("Binding outbound interface at " + _bindAddress);
 
-                InetSocketAddress bindAddr = new InetSocketAddress(_bindAddress, 0);
+                final InetSocketAddress bindAddr = new InetSocketAddress(_bindAddress, 0);
                 _clientConnection.socket().bind(bindAddr);
             }
 
-            InetSocketAddress peerAddr = new InetSocketAddress(_host, _port);
+            final InetSocketAddress peerAddr = new InetSocketAddress(_host, _port);
             _clientConnection.connect(peerAddr);
 
             SSLEngine sslEngine = null;
             // Begin SSL handshake in BLOCKING mode
             _clientConnection.configureBlocking(true);
 
-            SSLContext sslContext = Link.initSSLContext(true);
+            final SSLContext sslContext = Link.initSSLContext(true);
             sslEngine = sslContext.createSSLEngine(_host, _port);
             sslEngine.setUseClientMode(true);
             sslEngine.setEnabledProtocols(SSLUtils.getSupportedProtocols(sslEngine.getEnabledProtocols()));
@@ -83,32 +82,31 @@ protected void init() throws IOException {
             s_logger.info("Connected to " + _host + ":" + _port);
 
             _clientConnection.configureBlocking(false);
-            Link link = new Link(peerAddr, this);
+            final Link link = new Link(peerAddr, this);
             link.setSSLEngine(sslEngine);
-            SelectionKey key = _clientConnection.register(_selector, SelectionKey.OP_READ);
+            final SelectionKey key = _clientConnection.register(_selector, SelectionKey.OP_READ);
             link.setKey(key);
             key.attach(link);
             // Notice we've already connected due to the handshake, so let's get the
             // remaining task done
             task = _factory.create(Task.Type.CONNECT, link, null);
-        } catch (GeneralSecurityException e) {
+        } catch (final GeneralSecurityException e) {
             _selector.close();
             throw new IOException("Failed to initialise security", e);
-        } catch (IOException e) {
+        } catch (final IOException e) {
             _selector.close();
             throw e;
         }
-
-        _executor.execute(task);
+        _executor.submit(task);
     }
 
     @Override
-    protected void registerLink(InetSocketAddress saddr, Link link) {
+    protected void registerLink(final InetSocketAddress saddr, final Link link) {
         // don't do anything.
     }
 
     @Override
-    protected void unregisterLink(InetSocketAddress saddr) {
+    protected void unregisterLink(final InetSocketAddress saddr) {
         // don't do anything.
     }
 
@@ -119,7 +117,5 @@ public void cleanUp() throws IOException {
             _clientConnection.close();
         }
         s_logger.info("NioClient connection closed");
-
     }
-
-}
+}