SAMZA-1568: Handle ZkInterruptedException in zkclient.close.

When zookeeper session failures occur in a stream processor,   leaves the group(zkClient is closed) and joins the group again.

The last step in that shutdown sequence is zkClient.close(). In some scenarios, it throws the following exception,

    org.I0Itec.zkclient.exception.ZkInterruptedException: java.lang.InterruptedException
    at org.I0Itec.zkclient.ZkClient.close(ZkClient.java:1278)
    at org.apache.samza.zk.ZkControllerImpl.stop(ZkControllerImpl.java:92)

    at org.apache.samza.zk.ZkJobCoordinator.stop(ZkJobCoordinator.java:141)
In existing implementation this is not handled, there by killing the stream processor.  The following codepath triggers this exception:

`StreamProcessor.stop -> ZkJobCoordinator.stop() ->  zkController.stop() -> zkUtils.close`

This exception causes the integration test to fail occasionally  and can cause LocalApplicationRunner.waitForFinish method call to block indefinitely(since this callback event success, updates the latch state required for waitForFinish to end).

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes apache#416 from shanthoosh/zk_utils_close

Author: shanthoosh

samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java

@@ -89,7 +89,7 @@ public void stop() {
 
     // close zk connection
     if (zkUtils != null) {
-      zkUtils.getZkClient().close();
+      zkUtils.close();
     }
   }
 

samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java

@@ -94,8 +94,6 @@ public int getGeneration() {
     return currentGeneration.get();
   }
 
-
-
   public ZkUtils(ZkKeyBuilder zkKeyBuilder, ZkClient zkClient, int connectionTimeoutMs, MetricsRegistry metricsRegistry) {
     this.keyBuilder = zkKeyBuilder;
     this.connectionTimeoutMs = connectionTimeoutMs;
@@ -298,7 +296,13 @@ public boolean exists(String path) {
   }
 
   public void close() throws ZkInterruptedException {
-    zkClient.close();
+    try {
+      zkClient.close();
+    } catch (ZkInterruptedException e) {
+      // Swallowing due to occurrence in the last stage of lifecycle (Not actionable) and clear the interrupted status.
+      Thread.interrupted();
+      LOG.warn("Ignoring the exception when closing the zookeeper client.", e);
+    }
   }
 
   /**

samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java

@@ -30,6 +30,7 @@
 import org.I0Itec.zkclient.IZkDataListener;
 import org.I0Itec.zkclient.ZkClient;
 import org.I0Itec.zkclient.ZkConnection;
+import org.I0Itec.zkclient.exception.ZkInterruptedException;
 import org.I0Itec.zkclient.exception.ZkNodeExistsException;
 import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.samza.SamzaException;
@@ -46,6 +47,7 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.mockito.Mockito;
 
 
 public class TestZkUtils {
@@ -88,8 +90,9 @@ public void testSetup() {
 
   @After
   public void testTeardown() {
-    zkUtils.close();
-    zkClient.close();
+    if (zkClient != null) {
+      zkUtils.close();
+    }
   }
 
   private ZkUtils getZkUtils() {
@@ -392,6 +395,15 @@ public int compare(String o1, String o2) {
 
   }
 
+  @Test
+  public void testCloseShouldNotThrowZkInterruptedExceptionToCaller() {
+    ZkClient zkClient = Mockito.mock(ZkClient.class);
+    ZkUtils zkUtils = new ZkUtils(KEY_BUILDER, zkClient,
+            SESSION_TIMEOUT_MS, new NoOpMetricsRegistry());
+    Mockito.doThrow(new ZkInterruptedException(new InterruptedException())).when(zkClient).close();
+    zkUtils.close();
+  }
+
   public static boolean testWithDelayBackOff(BooleanSupplier cond, long startDelayMs, long maxDelayMs) {
     long delay = startDelayMs;
     while (delay < maxDelayMs) {