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

kaaproject · Sep 6, 2016 · 0f7cca8 · 0f7cca8
1 parent da29bc9
commit 0f7cca8
Show file tree

Hide file tree

Showing 10 changed files with 114 additions and 5 deletions.
diff --git a/server/node/src/test/java/org/kaaproject/kaa/server/node/KaaNodeServerLauncherIT.java b/server/node/src/test/java/org/kaaproject/kaa/server/node/KaaNodeServerLauncherIT.java
@@ -27,6 +27,7 @@
 import org.apache.thrift.transport.TTransport;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.kaaproject.kaa.server.common.nosql.mongo.dao.MongoDBTestRunner;
 import org.kaaproject.kaa.server.common.thrift.KaaThriftService;
@@ -133,6 +134,7 @@ public void run() {
      *
      * @throws Exception the exception
      */
+    @Ignore("KAA-1281 Kaa node should block startup process if zookeeper is unavailable.")
     @Test
     public void testStartKaaNodeServerApplicationWithoutZkStarted() throws Exception {
         TTransport transport = null;

diff --git a/...g/kaaproject/kaa/server/node/service/initialization/KaaNodeInitializationServiceTest.java b/...g/kaaproject/kaa/server/node/service/initialization/KaaNodeInitializationServiceTest.java
@@ -17,7 +17,9 @@
 package org.kaaproject.kaa.server.node.service.initialization;
 
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.stubVoid;
 
+import org.apache.curator.framework.CuratorFramework;
 import org.apache.thrift.TMultiplexedProcessor;
 import org.apache.thrift.server.TThreadPoolServer;
 import org.apache.thrift.transport.TServerSocket;
@@ -36,6 +38,7 @@ public class KaaNodeInitializationServiceTest {
     private InitializationService controlInitializationService;
     private InitializationService bootstrapInitializationService;
     private InitializationService operationsInitializationService;
+    private CuratorFramework zkClient;
 
     /**
      * Test kaa node initialization service start.
@@ -119,14 +122,14 @@ public void testKaaNodeInitializationServiceStop() throws Exception {
     private KaaNodeInitializationService kaaNodeInitializationServiceSpy() throws Exception {
 
         KaaNodeInitializationService kaaNodeInitializationService = Mockito.spy(new KaaNodeInitializationService());
-        
+
         KaaNodeServerConfig kaaNodeServerConfig = new KaaNodeServerConfig();
         kaaNodeServerConfig.setThriftHost("localhost");
         kaaNodeServerConfig.setThriftPort(10090);        
         kaaNodeServerConfig.setControlServiceEnabled(true);
         kaaNodeServerConfig.setBootstrapServiceEnabled(true);
         kaaNodeServerConfig.setOperationsServiceEnabled(true);
-        
+
         ReflectionTestUtils.setField(kaaNodeInitializationService, "kaaNodeServerConfig", kaaNodeServerConfig);
 
         KaaNodeThriftService.Iface kaaNodeThriftService = Mockito.mock(KaaNodeThriftService.Iface.class);
@@ -141,13 +144,18 @@ private KaaNodeInitializationService kaaNodeInitializationServiceSpy() throws Ex
         controlInitializationService = mock(InitializationService.class);
         bootstrapInitializationService = mock(InitializationService.class);
         operationsInitializationService = mock(InitializationService.class);
-
+        zkClient = mock(CuratorFramework.class);
+        stubVoid(zkClient).toReturn().on().start();
+        stubVoid(zkClient).toReturn().on().blockUntilConnected();
+
         ReflectionTestUtils.setField(kaaNodeInitializationService, "controlInitializationService", controlInitializationService);
 
         ReflectionTestUtils.setField(kaaNodeInitializationService, "bootstrapInitializationService", bootstrapInitializationService);
 
         ReflectionTestUtils.setField(kaaNodeInitializationService, "operationsInitializationService", operationsInitializationService);
-
+
+        ReflectionTestUtils.setField(kaaNodeInitializationService, "zkClient", zkClient);
+
         TServerSocket serverSocket = Mockito.mock(TServerSocket.class);
 
         Mockito.doReturn(serverSocket).when(kaaNodeInitializationService).createServerSocket();

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
@@ -48,6 +48,20 @@
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />
     </bean>
 
+    <!--Zookeeper client configuration-->
+
+    <bean id="zkRetryPolicy" class="org.apache.curator.retry.RetryUntilElapsed">
+        <constructor-arg name="maxElapsedTimeMs" value="#{properties[zk_max_retry_time]}"/>
+        <constructor-arg name="sleepMsBetweenRetries" value="#{properties[zk_sleep_time]}"/>
+    </bean>
+
+    <bean id="zkClient" class="org.apache.curator.framework.CuratorFrameworkFactory" factory-method="newClient">
+        <constructor-arg name="connectString" value="#{properties[zk_host_port_list]}"/>
+        <constructor-arg name="sessionTimeoutMs" value="60000"/>
+        <constructor-arg name="connectionTimeoutMs" value="3000"/>
+        <constructor-arg name="retryPolicy" ref="zkRetryPolicy"/>
+    </bean>
+
     <!-- Bootstrap Service -->
 
     <bean id="bootstrapInitializationService" class="org.kaaproject.kaa.server.bootstrap.service.initialization.BootstrapInitializationService" />

diff --git a/server/node/src/test/resources/common-test-context.xml b/server/node/src/test/resources/common-test-context.xml
@@ -56,6 +56,20 @@
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />
     </bean>
 
+    <!--Zookeeper client configuration-->
+
+    <bean id="zkRetryPolicy" class="org.apache.curator.retry.RetryUntilElapsed">
+        <constructor-arg name="maxElapsedTimeMs" value="#{properties[zk_max_retry_time]}"/>
+        <constructor-arg name="sleepMsBetweenRetries" value="#{properties[zk_sleep_time]}"/>
+    </bean>
+
+    <bean id="zkClient" class="org.apache.curator.framework.CuratorFrameworkFactory" factory-method="newClient">
+        <constructor-arg name="connectString" value="#{properties[zk_host_port_list]}"/>
+        <constructor-arg name="sessionTimeoutMs" value="60000"/>
+        <constructor-arg name="connectionTimeoutMs" value="3000"/>
+        <constructor-arg name="retryPolicy" ref="zkRetryPolicy"/>
+    </bean>
+
     <!-- Kaa Node Service -->
 
     <bean id="kaaNodeInitializationService"

diff --git a/server/node/src/test/resources/common-zk-test-context.xml b/server/node/src/test/resources/common-zk-test-context.xml
@@ -56,6 +56,20 @@
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />
     </bean>
 
+    <!--Zookeeper client configuration-->
+
+    <bean id="zkRetryPolicy" class="org.apache.curator.retry.RetryUntilElapsed">
+        <constructor-arg name="maxElapsedTimeMs" value="#{properties[zk_max_retry_time]}"/>
+        <constructor-arg name="sleepMsBetweenRetries" value="#{properties[zk_sleep_time]}"/>
+    </bean>
+
+    <bean id="zkClient" class="org.apache.curator.framework.CuratorFrameworkFactory" factory-method="newClient">
+        <constructor-arg name="connectString" value="#{properties[zk_host_port_list]}"/>
+        <constructor-arg name="sessionTimeoutMs" value="60000"/>
+        <constructor-arg name="connectionTimeoutMs" value="3000"/>
+        <constructor-arg name="retryPolicy" ref="zkRetryPolicy"/>
+    </bean>
+
     <!-- Kaa Node Service -->
 
     <bean id="kaaNodeInitializationService"

diff --git a/server/node/src/test/resources/control/common-test-context.xml b/server/node/src/test/resources/control/common-test-context.xml
@@ -54,6 +54,20 @@
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />
     </bean>
 
+    <!--Zookeeper client configuration-->
+
+    <bean id="zkRetryPolicy" class="org.apache.curator.retry.RetryUntilElapsed">
+        <constructor-arg name="maxElapsedTimeMs" value="#{properties[zk_max_retry_time]}"/>
+        <constructor-arg name="sleepMsBetweenRetries" value="#{properties[zk_sleep_time]}"/>
+    </bean>
+
+    <bean id="zkClient" class="org.apache.curator.framework.CuratorFrameworkFactory" factory-method="newClient">
+        <constructor-arg name="connectString" value="#{properties[zk_host_port_list]}"/>
+        <constructor-arg name="sessionTimeoutMs" value="60000"/>
+        <constructor-arg name="connectionTimeoutMs" value="3000"/>
+        <constructor-arg name="retryPolicy" ref="zkRetryPolicy"/>
+    </bean>
+
     <!-- Kaa Node Service -->
 
     <bean id="kaaNodeInitializationService"

diff --git a/server/node/src/test/resources/operations/cache-test-context.xml b/server/node/src/test/resources/operations/cache-test-context.xml
@@ -42,6 +42,21 @@
     <context:component-scan base-package="org.kaaproject.kaa.server.operations.service.cache.concurrent" />
 
     <util:properties id="properties" location="classpath:/operations/operations-server-test.properties" />
+
+    <!--Zookeeper client configuration-->
+
+    <bean id="zkRetryPolicy" class="org.apache.curator.retry.RetryUntilElapsed">
+        <constructor-arg name="maxElapsedTimeMs" value="#{properties[zk_max_retry_time]}"/>
+        <constructor-arg name="sleepMsBetweenRetries" value="#{properties[zk_sleep_time]}"/>
+    </bean>
+
+    <bean id="zkClient" class="org.apache.curator.framework.CuratorFrameworkFactory" factory-method="newClient">
+        <constructor-arg name="connectString" value="#{properties[zk_host_port_list]}"/>
+        <constructor-arg name="sessionTimeoutMs" value="60000"/>
+        <constructor-arg name="connectionTimeoutMs" value="3000"/>
+        <constructor-arg name="retryPolicy" ref="zkRetryPolicy"/>
+    </bean>
+
     <bean id="cacheService" class="org.kaaproject.kaa.server.operations.service.cache.concurrent.ConcurrentCacheService"/>
 
     <bean id="ehcache" class="org.springframework.cache.ehcache.EhCacheManagerFactoryBean" p:config-location="ehcache.xml"/>

diff --git a/server/node/src/test/resources/operations/common-test-context.xml b/server/node/src/test/resources/operations/common-test-context.xml
@@ -52,6 +52,20 @@
         <property name="zkIgnoreErrors" value="#{properties[zk_ignore_errors]}" />
     </bean>
 
+    <!--Zookeeper client configuration-->
+
+    <bean id="zkRetryPolicy" class="org.apache.curator.retry.RetryUntilElapsed">
+        <constructor-arg name="maxElapsedTimeMs" value="#{properties[zk_max_retry_time]}"/>
+        <constructor-arg name="sleepMsBetweenRetries" value="#{properties[zk_sleep_time]}"/>
+    </bean>
+
+    <bean id="zkClient" class="org.apache.curator.framework.CuratorFrameworkFactory" factory-method="newClient">
+        <constructor-arg name="connectString" value="#{properties[zk_host_port_list]}"/>
+        <constructor-arg name="sessionTimeoutMs" value="60000"/>
+        <constructor-arg name="connectionTimeoutMs" value="3000"/>
+        <constructor-arg name="retryPolicy" ref="zkRetryPolicy"/>
+    </bean>
+
     <!-- Operations Service -->
 
     <bean id="operationsInitializationService" class="org.kaaproject.kaa.server.operations.service.initialization.OperationsInitializationService" />

diff --git a/server/node/src/test/resources/operations/config-test-context.xml b/server/node/src/test/resources/operations/config-test-context.xml
@@ -38,7 +38,7 @@
 
     <util:properties id="properties"
         location="classpath:/operations/config-server-test.properties" />
-	
+
     <bean id="operationsInitializationService" class="org.kaaproject.kaa.server.operations.service.event.ESTestOperationsInitializationService" />
     <bean id="akkaService" class="org.kaaproject.kaa.server.operations.service.event.ESTestAkkaService" />
     <bean id="operationsService" class="org.kaaproject.kaa.server.operations.service.event.ESTestOperationsService" />

diff --git a/server/node/src/test/resources/operations/operations-server-eventService-test-IT.xml b/server/node/src/test/resources/operations/operations-server-eventService-test-IT.xml
@@ -34,6 +34,20 @@
     <context:component-scan base-package="org.kaaproject.kaa.server.operations.service.event" />
     <util:properties id="properties" location="classpath:/operations/operations-server-zk-test.properties" />
 
+    <!--Zookeeper client configuration-->
+
+    <bean id="zkRetryPolicy" class="org.apache.curator.retry.RetryUntilElapsed">
+        <constructor-arg name="maxElapsedTimeMs" value="#{properties[zk_max_retry_time]}"/>
+        <constructor-arg name="sleepMsBetweenRetries" value="#{properties[zk_sleep_time]}"/>
+    </bean>
+
+    <bean id="zkClient" class="org.apache.curator.framework.CuratorFrameworkFactory" factory-method="newClient">
+        <constructor-arg name="connectString" value="#{properties[zk_host_port_list]}"/>
+        <constructor-arg name="sessionTimeoutMs" value="60000"/>
+        <constructor-arg name="connectionTimeoutMs" value="3000"/>
+        <constructor-arg name="retryPolicy" ref="zkRetryPolicy"/>
+    </bean>
+
     <bean id="operationsInitializationService" class="org.kaaproject.kaa.server.operations.service.event.ESTestOperationsInitializationService" />
     <bean id="cacheService" class="org.kaaproject.kaa.server.operations.service.event.ESTestCacheService" />
     <bean id="akkaService" class="org.kaaproject.kaa.server.operations.service.event.ESTestAkkaService" />