KAA-1281 : block kaa-node startup if zookeeper is unavailable

kaaproject · Sep 7, 2016 · 3eb7232 · 3eb7232
1 parent 0f7cca8
commit 3eb7232
Show file tree

Hide file tree

Showing 18 changed files with 89 additions and 18 deletions.
diff --git a/...rver-shared/src/main/java/org/kaaproject/kaa/server/common/AbstractServerApplication.java b/...rver-shared/src/main/java/org/kaaproject/kaa/server/common/AbstractServerApplication.java
@@ -118,7 +118,7 @@ public void startAndWait(String[] args) {
         }
 
         LOG.info("{} application stopped.", getName());
-    };
+    }
 
     /**
      * Gets the name of the service.

diff --git a/...node/src/main/java/org/kaaproject/kaa/server/node/service/config/KaaNodeServerConfig.java b/...node/src/main/java/org/kaaproject/kaa/server/node/service/config/KaaNodeServerConfig.java
@@ -42,6 +42,9 @@ public class KaaNodeServerConfig {
     /** The zk host port list. */
     private String zkHostPortList;
 
+    /** Time to connect to ZK. */
+    private int zkWaitConnectionTime;
+
     /** The zk max retry time. */
     private int zkMaxRetryTime;
 
@@ -106,6 +109,15 @@ public String getZkHostPortList() {
         return zkHostPortList;
     }
 
+    /**
+     * Time to connect to ZK.
+     *
+     * @return the zkWaitConnectionTime
+     */
+    public int getZkWaitConnectionTime() {
+        return zkWaitConnectionTime;
+    }
+
     /**
      * The zk max retry time.
      * 
@@ -189,6 +201,14 @@ public void setZkHostPortList(String zkHostPortList) {
         this.zkHostPortList = zkHostPortList;
     }
 
+    /**
+     * @param zkWaitConnectionTime
+     *            the zkWaitConnectionTime to set
+     */
+    public void setZkWaitConnectionTime(int zkWaitConnectionTime) {
+        this.zkWaitConnectionTime = zkWaitConnectionTime;
+    }
+
     /**
      * @param zkMaxRetryTime
      *            the zkMaxRetryTime to set

diff --git a/...a/org/kaaproject/kaa/server/node/service/initialization/KaaNodeInitializationService.java b/...a/org/kaaproject/kaa/server/node/service/initialization/KaaNodeInitializationService.java
@@ -39,6 +39,7 @@
 import org.apache.thrift.transport.TSocket;
 import org.apache.thrift.transport.TTransportException;
 import org.kaaproject.kaa.server.common.thrift.KaaThriftService;
+import org.kaaproject.kaa.server.common.thrift.cli.client.BaseCliThriftClient;
 import org.kaaproject.kaa.server.common.thrift.gen.bootstrap.BootstrapThriftService;
 import org.kaaproject.kaa.server.common.thrift.gen.node.KaaNodeThriftService;
 import org.kaaproject.kaa.server.common.thrift.gen.operations.OperationsThriftService;
@@ -104,20 +105,23 @@ public void start() {
             LOG.error("Interrupted while waiting for thrift to start...", e);
         }
 
-        waitZkConnection();
+        if (waitZkConnection()) {
+            if (getNodeConfig().isControlServiceEnabled()) {
+                controlInitializationService.start();
+            }
+            if (getNodeConfig().isBootstrapServiceEnabled()) {
+                bootstrapInitializationService.start();
+            }
+            if (getNodeConfig().isOperationsServiceEnabled()) {
+                operationsInitializationService.start();
+            }
 
-        if (getNodeConfig().isControlServiceEnabled()) {
+            LOG.info("Kaa Node Server Started.");
-            controlInitializationService.start();
+        } else {
-        }
+            LOG.error("Failed to connect to Zookeeper within {} minutes. Kaa Node Server will be stopped.", getNodeConfig().getZkWaitConnectionTime());
-        if (getNodeConfig().isBootstrapServiceEnabled()) {
+            stopThrift();
-            bootstrapInitializationService.start();
         }
-        if (getNodeConfig().isOperationsServiceEnabled()) {
+
-            operationsInitializationService.start();
-        }
-
-        LOG.info("Kaa Node Server Started.");
-
         try {
             thriftShutdownLatch.await();
         } catch (InterruptedException e) {
@@ -126,15 +130,16 @@ public void start() {
 
     }
 
-    private void waitZkConnection() {
+    private boolean waitZkConnection() {
         if (!zkClient.isStarted()) {
             zkClient.start();
         }
         try {
             LOG.info("Waiting connection to Zookeeper at ", getNodeConfig().getZkHostPortList());
-            zkClient.blockUntilConnected();
+            return zkClient.blockUntilConnected(getNodeConfig().getZkWaitConnectionTime(), TimeUnit.MINUTES);
         } catch (InterruptedException e) {
             LOG.error("Zookeeper client was interrupted while waiting for connection! ", getNodeConfig().getZkHostPortList(), e);
+            return false;
         }
     }
 
@@ -162,6 +167,22 @@ public void stop() {
         LOG.info("Kaa Node Server Stopped.");
     }
 
+    /**
+     * Stop thrift.
+     *
+     */
+    public void stopThrift() {
+        if (zkClient!=null) {
+            zkClient.close();
+        }
+        if (server!=null) {
+            server.stop();
+        }
+        ThriftExecutor.shutdown();
+
+        LOG.info("Kaa Node Server Stopped.");
+    }
+
     /**
      * Start thrift.
      *

diff --git a/...e/src/main/java/org/kaaproject/kaa/server/operations/service/akka/DefaultAkkaService.java b/...e/src/main/java/org/kaaproject/kaa/server/operations/service/akka/DefaultAkkaService.java
@@ -160,6 +160,9 @@ public void onNotification(Notification notification) {
     @PreDestroy
     public void preDestroy() {
         context.getEventService().removeListener(eventListener);
+        akka.stop(ioRouter);
+        akka.stop(opsActor);
+        akka.terminate();
     }
 
     @Override

diff --git a/server/node/src/main/resources/kaa-node.properties b/server/node/src/main/resources/kaa-node.properties
@@ -33,6 +33,9 @@ zk_enabled=true
 # Zookeeper service url list.
 zk_host_port_list=localhost:2181
 
+# Wait for connection time in minutes.
+zk_wait_connection_time=5
+
 # The max retry time in milliseconds.
 zk_max_retry_time=3000
 

diff --git a/server/node/src/main/resources/kaaNodeContext.xml b/server/node/src/main/resources/kaaNodeContext.xml
@@ -54,6 +54,7 @@
         <property name="thriftPort" value="#{properties[thrift_port]}" />
         <property name="zkEnabled" value="#{properties[zk_enabled]}" />
         <property name="zkHostPortList" value="#{properties[zk_host_port_list]}" />
+        <property name="zkWaitConnectionTime" value="#{properties[zk_wait_connection_time]}" />
         <property name="zkMaxRetryTime" value="#{properties[zk_max_retry_time]}" />
         <property name="zkSleepTime" value="#{properties[zk_sleep_time]}" />
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />

diff --git a/...g/kaaproject/kaa/server/node/service/initialization/KaaNodeInitializationServiceTest.java b/...g/kaaproject/kaa/server/node/service/initialization/KaaNodeInitializationServiceTest.java
@@ -16,8 +16,11 @@
 
 package org.kaaproject.kaa.server.node.service.initialization;
 
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.stubVoid;
+import static org.mockito.Mockito.when;
 
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.thrift.TMultiplexedProcessor;
@@ -30,9 +33,12 @@
 import org.kaaproject.kaa.server.common.thrift.gen.node.KaaNodeThriftService;
 import org.kaaproject.kaa.server.common.thrift.gen.operations.OperationsThriftService;
 import org.kaaproject.kaa.server.node.service.config.KaaNodeServerConfig;
+import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import java.util.concurrent.TimeUnit;
+
 public class KaaNodeInitializationServiceTest {
 
     private InitializationService controlInitializationService;
@@ -129,6 +135,7 @@ private KaaNodeInitializationService kaaNodeInitializationServiceSpy() throws Ex
         kaaNodeServerConfig.setControlServiceEnabled(true);
         kaaNodeServerConfig.setBootstrapServiceEnabled(true);
         kaaNodeServerConfig.setOperationsServiceEnabled(true);
+        kaaNodeServerConfig.setZkWaitConnectionTime(5);
 
         ReflectionTestUtils.setField(kaaNodeInitializationService, "kaaNodeServerConfig", kaaNodeServerConfig);
 
@@ -145,8 +152,9 @@ private KaaNodeInitializationService kaaNodeInitializationServiceSpy() throws Ex
         bootstrapInitializationService = mock(InitializationService.class);
         operationsInitializationService = mock(InitializationService.class);
         zkClient = mock(CuratorFramework.class);
-        stubVoid(zkClient).toReturn().on().start();
+        doNothing().when(zkClient).start();
-        stubVoid(zkClient).toReturn().on().blockUntilConnected();
+        doNothing().when(zkClient).blockUntilConnected();
+        when(zkClient.blockUntilConnected(anyInt(), any(TimeUnit.class))).thenReturn(true);
 
         ReflectionTestUtils.setField(kaaNodeInitializationService, "controlInitializationService", controlInitializationService);
 

diff --git a/server/node/src/test/resources/bootstrap/bootstrap-server-test.properties b/server/node/src/test/resources/bootstrap/bootstrap-server-test.properties
@@ -2,6 +2,7 @@ thrift_host=localhost
 thrift_port=19092
 zk_enabled=true
 zk_host_port_list=localhost:2185
+zk_wait_connection_time=5
 zk_max_retry_time=3000
 zk_sleep_time=1000
 zk_ignore_errors=true

diff --git a/server/node/src/test/resources/bootstrap/common-zk-test-context.xml b/server/node/src/test/resources/bootstrap/common-zk-test-context.xml
@@ -43,6 +43,7 @@
         <property name="thriftPort" value="#{properties[thrift_port]}" />
         <property name="zkEnabled" value="#{properties[zk_enabled]}" />
         <property name="zkHostPortList" value="#{properties[zk_host_port_list]}" />
+        <property name="zkWaitConnectionTime" value="#{properties[zk_wait_connection_time]}" />
         <property name="zkMaxRetryTime" value="#{properties[zk_max_retry_time]}" />
         <property name="zkSleepTime" value="#{properties[zk_sleep_time]}" />
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />

diff --git a/server/node/src/test/resources/common-test-context.xml b/server/node/src/test/resources/common-test-context.xml
@@ -51,6 +51,7 @@
         <property name="thriftPort" value="#{properties[thrift_port]}" />
         <property name="zkEnabled" value="#{properties[zk_enabled]}" />
         <property name="zkHostPortList" value="#{properties[zk_host_port_list]}" />
+        <property name="zkWaitConnectionTime" value="#{properties[zk_wait_connection_time]}" />
         <property name="zkMaxRetryTime" value="#{properties[zk_max_retry_time]}" />
         <property name="zkSleepTime" value="#{properties[zk_sleep_time]}" />
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />

diff --git a/server/node/src/test/resources/common-zk-test-context.xml b/server/node/src/test/resources/common-zk-test-context.xml
@@ -51,6 +51,7 @@
         <property name="thriftPort" value="#{properties[thrift_port]}" />
         <property name="zkEnabled" value="#{properties[zk_enabled]}" />
         <property name="zkHostPortList" value="#{properties[zk_host_port_list]}" />
+        <property name="zkWaitConnectionTime" value="#{properties[zk_wait_connection_time]}" />
         <property name="zkMaxRetryTime" value="#{properties[zk_max_retry_time]}" />
         <property name="zkSleepTime" value="#{properties[zk_sleep_time]}" />
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />

diff --git a/server/node/src/test/resources/control/common-test-context.xml b/server/node/src/test/resources/control/common-test-context.xml
@@ -49,6 +49,7 @@
         <property name="thriftPort" value="#{properties[thrift_port]}" />
         <property name="zkEnabled" value="#{properties[zk_enabled]}" />
         <property name="zkHostPortList" value="#{properties[zk_host_port_list]}" />
+        <property name="zkWaitConnectionTime" value="#{properties[zk_wait_connection_time]}" />
         <property name="zkMaxRetryTime" value="#{properties[zk_max_retry_time]}" />
         <property name="zkSleepTime" value="#{properties[zk_sleep_time]}" />
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />

diff --git a/server/node/src/test/resources/control/control-server-test.properties b/server/node/src/test/resources/control/control-server-test.properties
@@ -9,6 +9,7 @@ admin_port=9080
 
 zk_enabled=true
 zk_host_port_list=localhost:2185
+zk_wait_connection_time=5
 zk_max_retry_time=3000
 zk_sleep_time=1000
 zk_ignore_errors=true

diff --git a/server/node/src/test/resources/kaa-node-test.properties b/server/node/src/test/resources/kaa-node-test.properties
@@ -8,6 +8,7 @@ thrift_port=10090
 admin_port=9080
 zk_enabled=true
 zk_host_port_list=localhost:2185
+zk_wait_connection_time=5
 zk_max_retry_time=3000
 zk_sleep_time=1000
 zk_ignore_errors=true

diff --git a/server/node/src/test/resources/kaa-node-zk-test.properties b/server/node/src/test/resources/kaa-node-zk-test.properties
@@ -8,6 +8,7 @@ thrift_port=10090
 admin_port=9080
 zk_enabled=false
 zk_host_port_list=localhost:2185
+zk_wait_connection_time=5
 zk_max_retry_time=3000
 zk_sleep_time=1000
 zk_ignore_errors=true

diff --git a/server/node/src/test/resources/operations/common-test-context.xml b/server/node/src/test/resources/operations/common-test-context.xml
@@ -47,6 +47,7 @@
         <property name="thriftPort" value="#{properties[thrift_port]}" />
         <property name="zkEnabled" value="#{properties[zk_enabled]}" />
         <property name="zkHostPortList" value="#{properties[zk_host_port_list]}" />
+        <property name="zkWaitConnectionTime" value="#{properties[zk_wait_connection_time]}" />
         <property name="zkMaxRetryTime" value="#{properties[zk_max_retry_time]}" />
         <property name="zkSleepTime" value="#{properties[zk_sleep_time]}" />
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />

diff --git a/server/node/src/test/resources/operations/operations-server-test.properties b/server/node/src/test/resources/operations/operations-server-test.properties
@@ -19,6 +19,9 @@ zk_enabled=false
 # Zookeeper service url list
 zk_host_port_list=localhost:2181
 
+# Wait for connection time in milliseconds.
+zk_wait_connection_time=5
+
 # The max retry time in milliseconds
 zk_max_retry_time=3000
 

diff --git a/server/node/src/test/resources/operations/operations-server-zk-test.properties b/server/node/src/test/resources/operations/operations-server-zk-test.properties
@@ -19,6 +19,9 @@ zk_enabled=true
 # Zookeeper service url list
 zk_host_port_list=localhost:2185
 
+# Wait for connection time in milliseconds.
+zk_wait_connection_time=5
+
 # The max retry time in milliseconds
 zk_max_retry_time=3000