SAMZA-1336. Session disconnect propagation.
authorBoris Shkolnik <boryas@apache.org>
Mon, 17 Jul 2017 22:57:29 +0000 (15:57 -0700)
committernavina <navina@apache.org>
Mon, 17 Jul 2017 22:57:29 +0000 (15:57 -0700)
If ZK doesn't receive any communication from a zkClient (including heartbeats), it closes the session with the client. It removes all the ephemeral nodes associated with the client. That's why we need to restore all these nodes - need to re-register.
We are using ZkClient library to connect to zookeeper. This library allows us to get notification when the session is closed and when a new session is created. So when the new session is created we reset all session related state and re-register.
One weird feature of the library/zookeeper is that when a new session is established, it is still possible to receive old notifications. To avoid this we introduced generation number which we pass into each callback. And if the generation number has changed when the callback was invoked, we will ignore this callback.

Author: Boris Shkolnik <boryas@apache.org>
Author: Boris Shkolnik <bshkolni@bshkolni-ld1.linkedin.biz>
Author: Boris Shkolnik <bshkolni@linkedin.com>

Reviewers: Navina Ramesh <navina@apache.org>, Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Closes #229 from sborya/SessionFailReregister

21 files changed:
docs/learn/documentation/versioned/jobs/configuration-table.html
samza-autoscaling/src/main/java/org/apache/samza/autoscaling/deployer/ConfigManager.java
samza-core/src/main/java/org/apache/samza/config/TaskConfigJava.java
samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupByPartition.java
samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java
samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
samza-core/src/main/java/org/apache/samza/zk/ZkCoordinationServiceFactory.java
samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java
samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
samza-core/src/main/scala/org/apache/samza/config/JobConfig.scala
samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala
samza-hdfs/src/test/java/org/apache/samza/system/hdfs/reader/TestMultiFileHdfsReader.java
samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemConsumer.scala
samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngine.scala
samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessor.java [moved from samza-test/src/test/java/org/apache/samza/test/processor/TestZkStreamProcessor.java with 75% similarity]
samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorBase.java
samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorFailures.java [moved from samza-test/src/test/java/org/apache/samza/test/processor/TestZkStreamProcessorFailures.java with 84% similarity]
samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorSession.java [new file with mode: 0644]

index 7a1b56e..3263856 100644 (file)
 
                 <tr>
                     <td class="property" id="task-shutdown-ms">task.shutdown.ms</td>
-                    <td class="default">5000</td>
+                    <td class="default">30000</td>
                     <td class="description">
                         This property controls how long the Samza container will wait for an orderly shutdown of task instances.
                     </td>
index fd1e039..223d1d6 100644 (file)
@@ -178,7 +178,7 @@ public class ConfigManager {
   }
 
   /**
-   * skip all the unread messages up to the time this function is called.
+   * notAValidEvent all the unread messages up to the time this function is called.
    * This method just reads the messages, and it does not react to them or change any configuration of the system.
    */
   private void skipUnreadMessages() {
index f7b2bcd..fc9f165 100644 (file)
@@ -37,7 +37,7 @@ import scala.collection.JavaConverters;
 public class TaskConfigJava extends MapConfig {
   // Task Configs
   private static final String TASK_SHUTDOWN_MS = "task.shutdown.ms";
-  public static final long DEFAULT_TASK_SHUTDOWN_MS = 5000L;
+  public static final long DEFAULT_TASK_SHUTDOWN_MS = 30000L;
 
   // broadcast streams consumed by all tasks. e.g. kafka.foo#1
   public static final String BROADCAST_INPUT_STREAMS = "task.broadcast.inputs";
index 3022b72..8e3ef31 100644 (file)
@@ -56,7 +56,7 @@ public class GroupByPartition implements SystemStreamPartitionGrouper {
     Map<TaskName, Set<SystemStreamPartition>> groupedMap = new HashMap<TaskName, Set<SystemStreamPartition>>();
 
     for (SystemStreamPartition ssp : ssps) {
-      // skip the broadcast streams if there is any
+      // notAValidEvent the broadcast streams if there is any
       if (broadcastStreams.contains(ssp)) {
         continue;
       }
index 14a14a8..89edd16 100644 (file)
@@ -63,6 +63,7 @@ public class StreamProcessor {
   private final Map<String, MetricsReporter> customMetricsReporter;
   private final Config config;
   private final long taskShutdownMs;
+  private final String processorId;
 
   private ExecutorService executorService;
 
@@ -71,7 +72,7 @@ public class StreamProcessor {
   
   // Latch used to synchronize between the JobCoordinator thread and the container thread, when the container is
   // stopped due to re-balancing
-  private volatile CountDownLatch jcContainerShutdownLatch = new CountDownLatch(1);
+  /* package private */volatile CountDownLatch jcContainerShutdownLatch;
   private volatile boolean processorOnStartCalled = false;
 
   @VisibleForTesting
@@ -110,6 +111,11 @@ public class StreamProcessor {
     this(config, customMetricsReporters, (Object) streamTaskFactory, processorListener, null);
   }
 
+  @Override
+  public String toString() {
+    return "Processor:" + processorId;
+  }
+
   /* package private */
   JobCoordinator getJobCoordinator() {
     return Util.
@@ -119,6 +125,11 @@ public class StreamProcessor {
         .getJobCoordinator(config);
   }
 
+  @VisibleForTesting
+  JobCoordinator getCurrentJobCoordinator() {
+    return jobCoordinator;
+  }
+
   StreamProcessor(Config config, Map<String, MetricsReporter> customMetricsReporters, Object taskFactory,
                   StreamProcessorLifecycleListener processorListener, JobCoordinator jobCoordinator) {
     this.taskFactory = taskFactory;
@@ -129,6 +140,8 @@ public class StreamProcessor {
     this.jobCoordinator = (jobCoordinator != null) ? jobCoordinator : getJobCoordinator();
     this.jobCoordinatorListener = createJobCoordinatorListener();
     this.jobCoordinator.setListener(jobCoordinatorListener);
+
+    processorId = this.jobCoordinator.getProcessorId();
   }
 
   /**
@@ -197,6 +210,7 @@ public class StreamProcessor {
   }
 
   JobCoordinatorListener createJobCoordinatorListener() {
+    final String pid = this.toString();
     return new JobCoordinatorListener() {
 
       @Override
@@ -206,9 +220,10 @@ public class StreamProcessor {
           if (SamzaContainerStatus.NOT_STARTED.equals(status) || SamzaContainerStatus.STARTED.equals(status)) {
             boolean shutdownComplete = false;
             try {
-              LOGGER.info("Shutting down container in onJobModelExpired.");
+              LOGGER.info("Shutting down container in onJobModelExpired for processor:" + pid);
               container.pause();
               shutdownComplete = jcContainerShutdownLatch.await(taskShutdownMs, TimeUnit.MILLISECONDS);
+              LOGGER.info("ShutdownComplete=" + shutdownComplete);
             } catch (IllegalContainerStateException icse) {
               // Ignored since container is not running
               LOGGER.info("Container was not running.", icse);
@@ -216,6 +231,7 @@ public class StreamProcessor {
             } catch (InterruptedException e) {
               LOGGER.warn("Container shutdown was interrupted!" + container.toString(), e);
             }
+            LOGGER.info("Shutting down container done for pid=" + pid + "; complete =" + shutdownComplete);
             if (!shutdownComplete) {
               LOGGER.warn("Container " + container.toString() + " may not have shutdown successfully. " +
                   "Stopping the processor.");
index c1343b1..196e431 100644 (file)
@@ -19,8 +19,6 @@
 
 package org.apache.samza.zk;
 
-import org.I0Itec.zkclient.IZkChildListener;
-import org.I0Itec.zkclient.IZkDataListener;
 import org.apache.commons.collections4.CollectionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -93,7 +91,7 @@ public class ZkBarrierForVersionUpgrade {
 
     // subscribe for participant's list changes
     LOG.info("Subscribing for child changes at " + barrierParticipantsPath);
-    zkUtils.subscribeChildChanges(barrierParticipantsPath, new ZkBarrierChangeHandler(version, participants));
+    zkUtils.subscribeChildChanges(barrierParticipantsPath, new ZkBarrierChangeHandler(version, participants, zkUtils));
 
     barrierListenerOptional.ifPresent(zkBarrierListener -> zkBarrierListener.onBarrierCreated(version));
   }
@@ -106,7 +104,7 @@ public class ZkBarrierForVersionUpgrade {
    */
   public void join(String version, String participantId) {
     String barrierDonePath = keyBuilder.getBarrierStatePath(version);
-    zkUtils.subscribeDataChanges(barrierDonePath, new ZkBarrierReachedHandler(barrierDonePath, version));
+    zkUtils.subscribeDataChanges(barrierDonePath, new ZkBarrierReachedHandler(barrierDonePath, version, zkUtils));
 
     // TODO: Handle ZkNodeExistsException - SAMZA-1304
     zkUtils.getZkClient().createPersistent(
@@ -128,17 +126,21 @@ public class ZkBarrierForVersionUpgrade {
    * Listener for changes to the list of participants. It is meant to be subscribed only by the creator of the barrier
    * node. It checks to see when the barrier is ready to be marked as completed.
    */
-  class ZkBarrierChangeHandler implements IZkChildListener {
+  class ZkBarrierChangeHandler extends ZkUtils.GenIZkChildListener {
     private final String barrierVersion;
     private final List<String> names;
 
-    public ZkBarrierChangeHandler(String barrierVersion, List<String> names) {
+    public ZkBarrierChangeHandler(String barrierVersion, List<String> names, ZkUtils zkUtils) {
+      super(zkUtils, "ZkBarrierChangeHandler");
       this.barrierVersion = barrierVersion;
       this.names = names;
     }
 
     @Override
     public void handleChildChange(String parentPath, List<String> currentChildren) {
+      if (notAValidEvent()) {
+        return;
+      }
       if (currentChildren == null) {
         LOG.info("Got ZkBarrierChangeHandler handleChildChange with null currentChildren");
         return;
@@ -162,11 +164,12 @@ public class ZkBarrierForVersionUpgrade {
    * Barrier state values are either DONE or TIMED_OUT. It only registers to receive on valid state change notification.
    * Once a valid state change notification is received, it will un-subscribe from further notifications.
    */
-  class ZkBarrierReachedHandler implements IZkDataListener {
+  class ZkBarrierReachedHandler extends ZkUtils.GenIZkDataListener {
     private final String barrierStatePath;
     private final String barrierVersion;
 
-    public ZkBarrierReachedHandler(String barrierStatePath, String version) {
+    public ZkBarrierReachedHandler(String barrierStatePath, String version, ZkUtils zkUtils) {
+      super(zkUtils, "ZkBarrierReachedHandler");
       this.barrierStatePath = barrierStatePath;
       this.barrierVersion = version;
     }
@@ -174,6 +177,9 @@ public class ZkBarrierForVersionUpgrade {
     @Override
     public void handleDataChange(String dataPath, Object data) {
       LOG.info("got notification about barrier " + barrierStatePath + "; done=" + data);
+      if (notAValidEvent())
+        return;
+
       zkUtils.unsubscribeDataChanges(barrierStatePath, this);
       barrierListenerOptional.ifPresent(
           zkBarrierListener -> zkBarrierListener.onBarrierStateChanged(barrierVersion, (State) data));
@@ -182,7 +188,9 @@ public class ZkBarrierForVersionUpgrade {
     @Override
     public void handleDataDeleted(String dataPath)
         throws Exception {
-      LOG.warn("barrier done got deleted at " + dataPath);
+      LOG.warn("barrier done node got deleted at " + dataPath);
+      if (notAValidEvent())
+        return;
     }
   }
 
index a8317a8..3af5042 100644 (file)
@@ -19,8 +19,6 @@
 
 package org.apache.samza.zk;
 
-import org.I0Itec.zkclient.IZkChildListener;
-import org.I0Itec.zkclient.IZkDataListener;
 import org.apache.samza.SamzaException;
 import org.apache.samza.coordinator.LeaderElector;
 import org.slf4j.Logger;
@@ -49,10 +47,8 @@ public class ZkControllerImpl implements ZkController {
   private void init() {
     ZkKeyBuilder keyBuilder = zkUtils.getKeyBuilder();
     zkUtils.makeSurePersistentPathsExists(
-        new String[]{
-            keyBuilder.getProcessorsPath(),
-            keyBuilder.getJobModelVersionPath(),
-            keyBuilder.getJobModelPathPrefix()});
+        new String[]{keyBuilder.getProcessorsPath(), keyBuilder.getJobModelVersionPath(), keyBuilder
+            .getJobModelPathPrefix()});
   }
 
   @Override
@@ -62,7 +58,7 @@ public class ZkControllerImpl implements ZkController {
     zkLeaderElector.tryBecomeLeader();
 
     // subscribe to JobModel version updates
-    zkUtils.subscribeToJobModelVersionChange(new ZkJobModelVersionChangeHandler());
+    zkUtils.subscribeToJobModelVersionChange(new ZkJobModelVersionChangeHandler(zkUtils));
   }
 
   @Override
@@ -84,11 +80,16 @@ public class ZkControllerImpl implements ZkController {
 
   @Override
   public void subscribeToProcessorChange() {
-    zkUtils.subscribeToProcessorChange(new ProcessorChangeHandler());
+    zkUtils.subscribeToProcessorChange(new ProcessorChangeHandler(zkUtils));
   }
 
   // Only by Leader
-  class ProcessorChangeHandler implements IZkChildListener {
+  class ProcessorChangeHandler extends ZkUtils.GenIZkChildListener {
+
+    public ProcessorChangeHandler(ZkUtils zkUtils) {
+      super(zkUtils, "ProcessorChangeHandler");
+    }
+
     /**
      * Called when the children of the given path changed.
      *
@@ -97,7 +98,11 @@ public class ZkControllerImpl implements ZkController {
      * @throws Exception
      */
     @Override
-    public void handleChildChange(String parentPath, List<String> currentChildren) throws Exception {
+    public void handleChildChange(String parentPath, List<String> currentChildren)
+        throws Exception {
+      if (notAValidEvent())
+        return;
+
       if (currentChildren == null) {
         // this may happen only in case of exception in ZK. It happens if the zkNode has been deleted.
         // So the notification will pass 'null' as the list of children. Exception should be visible in the logs.
@@ -106,19 +111,27 @@ public class ZkControllerImpl implements ZkController {
         return;
       }
       LOG.info(
-          "ZkControllerImpl::ProcessorChangeHandler::handleChildChange - Path: " + parentPath + "  Current Children: "
-              + currentChildren);
+          "ZkControllerImpl::ProcessorChangeHandler::handleChildChange - Path: " + parentPath +
+              "  Current Children: " + currentChildren);
       zkControllerListener.onProcessorChange(currentChildren);
+
     }
   }
 
-  class ZkJobModelVersionChangeHandler implements IZkDataListener {
+  class ZkJobModelVersionChangeHandler extends ZkUtils.GenIZkDataListener {
+
+    public ZkJobModelVersionChangeHandler(ZkUtils zkUtils) {
+      super(zkUtils, "ZkJobModelVersionChangeHandler");
+    }
     /**
      * Called when there is a change to the data in JobModel version path
      * To the subscribers, it signifies that a new version of JobModel is available.
      */
     @Override
     public void handleDataChange(String dataPath, Object data) throws Exception {
+      if (notAValidEvent())
+        return;
+
       LOG.info("pid=" + processorIdStr + ". Got notification on version update change. path=" + dataPath + "; data="
           + data);
       zkControllerListener.onNewJobModelAvailable((String) data);
@@ -126,6 +139,9 @@ public class ZkControllerImpl implements ZkController {
 
     @Override
     public void handleDataDeleted(String dataPath) throws Exception {
+      if (notAValidEvent())
+        return;
+
       throw new SamzaException("version update path has been deleted!");
     }
   }
index d0633a8..1dd5ec9 100644 (file)
@@ -41,7 +41,6 @@ public class ZkCoordinationServiceFactory implements CoordinationServiceFactory
         createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 
     ZkUtils zkUtils = new ZkUtils(new ZkKeyBuilder(groupId), zkClient, zkConfig.getZkConnectionTimeoutMs(), new NoOpMetricsRegistry());
-
     return new ZkCoordinationUtils(participantId, zkConfig, zkUtils);
   }
 
index 8ca26c8..298c96e 100644 (file)
  */
 package org.apache.samza.zk;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import org.I0Itec.zkclient.IZkStateListener;
+import org.I0Itec.zkclient.ZkClient;
 import java.util.Set;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.samza.config.ApplicationConfig;
@@ -37,16 +40,18 @@ import org.apache.samza.coordinator.JobModelManager;
 import org.apache.samza.coordinator.LeaderElector;
 import org.apache.samza.coordinator.LeaderElectorListener;
 import org.apache.samza.job.model.JobModel;
-import org.apache.samza.metrics.MetricsReporter;
 import org.apache.samza.metrics.MetricsRegistry;
+import org.apache.samza.metrics.MetricsReporter;
 import org.apache.samza.metrics.ReadableMetricsRegistry;
 import org.apache.samza.runtime.ProcessorIdGenerator;
 import org.apache.samza.system.StreamMetadataCache;
 import org.apache.samza.util.ClassLoaderHelper;
 import org.apache.samza.util.MetricsReporterLoader;
+import org.apache.zookeeper.Watcher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
 /**
  * JobCoordinator for stand alone processor managed via Zookeeper.
  */
@@ -75,26 +80,27 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
     this.config = config;
     ZkConfig zkConfig = new ZkConfig(config);
     ZkKeyBuilder keyBuilder = new ZkKeyBuilder(new ApplicationConfig(config).getGlobalAppId());
+    ZkClient zkClient = ZkCoordinationServiceFactory
+        .createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
+
+    // setup a listener for a session state change
+    // we are mostly interested in "session closed" and "new session created" events
+    zkClient.subscribeStateChanges(new ZkSessionStateChangedListener());
+
     this.metrics = new ZkJobCoordinatorMetrics(metricsRegistry);
     this.zkUtils = new ZkUtils(
         keyBuilder,
-        ZkCoordinationServiceFactory.createZkClient(
-            zkConfig.getZkConnect(),
-            zkConfig.getZkSessionTimeoutMs(),
-            zkConfig.getZkConnectionTimeoutMs()),
+        zkClient,
         zkConfig.getZkConnectionTimeoutMs(), metricsRegistry);
 
     this.processorId = createProcessorId(config);
     LeaderElector leaderElector = new ZkLeaderElector(processorId, zkUtils);
     leaderElector.setLeaderElectorListener(new LeaderElectorListenerImpl());
     this.zkController = new ZkControllerImpl(processorId, zkUtils, this, leaderElector);
-    this.barrier =  new ZkBarrierForVersionUpgrade(
-        keyBuilder.getJobModelVersionBarrierPrefix(),
-        zkUtils,
+    this.barrier = new ZkBarrierForVersionUpgrade(keyBuilder.getJobModelVersionBarrierPrefix(), zkUtils,
         new ZkBarrierListenerImpl());
     this.debounceTimeMs = new JobConfig(config).getDebounceTimeMs();
     this.reporters = MetricsReporterLoader.getMetricsReporters(new MetricsConfig(config), processorId);
-
   }
 
   @Override
@@ -102,7 +108,8 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
     startMetrics();
     streamMetadataCache = StreamMetadataCache.apply(METADATA_CACHE_TTL_MS, config);
 
-    debounceTimer = new ScheduleAfterDebounceTime(throwable -> {
+    debounceTimer = new ScheduleAfterDebounceTime(throwable ->
+      {
         LOG.error("Received exception from in JobCoordinator Processing!", throwable);
         stop();
       });
@@ -158,8 +165,8 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
   @Override
   public void onProcessorChange(List<String> processors) {
     LOG.info("ZkJobCoordinator::onProcessorChange - list of processors changed! List size=" + processors.size());
-    debounceTimer.scheduleAfterDebounceTime(ScheduleAfterDebounceTime.ON_PROCESSOR_CHANGE,
-        debounceTimeMs, () -> doOnProcessorChange(processors));
+    debounceTimer.scheduleAfterDebounceTime(ScheduleAfterDebounceTime.ON_PROCESSOR_CHANGE, debounceTimeMs,
+        () -> doOnProcessorChange(processors));
   }
 
   void doOnProcessorChange(List<String> processors) {
@@ -176,7 +183,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
     // Generate the JobModel
     JobModel jobModel = generateNewJobModel(currentProcessorIds);
     // Assign the next version of JobModel
-    String currentJMVersion  = zkUtils.getJobModelVersion();
+    String currentJMVersion = zkUtils.getJobModelVersion();
     String nextJMVersion;
     if (currentJMVersion == null) {
       nextJMVersion = "1";
@@ -260,8 +267,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
    * Generate new JobModel when becoming a leader or the list of processor changed.
    */
   private JobModel generateNewJobModel(List<String> processors) {
-    return JobModelManager.readJobModel(this.config, Collections.emptyMap(), null, streamMetadataCache,
-        processors);
+    return JobModelManager.readJobModel(this.config, Collections.emptyMap(), null, streamMetadataCache, processors);
   }
 
   class LeaderElectorListenerImpl implements LeaderElectorListener {
@@ -270,9 +276,8 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
       LOG.info("ZkJobCoordinator::onBecomeLeader - I became the leader!");
       metrics.isLeader.set(true);
       zkController.subscribeToProcessorChange();
-      debounceTimer.scheduleAfterDebounceTime(
-        ScheduleAfterDebounceTime.ON_PROCESSOR_CHANGE,
-        debounceTimeMs, () -> {
+      debounceTimer.scheduleAfterDebounceTime(ScheduleAfterDebounceTime.ON_PROCESSOR_CHANGE, debounceTimeMs, () ->
+        {
           // actual actions to do are the same as onProcessorChange
           doOnProcessorChange(new ArrayList<>());
         });
@@ -281,6 +286,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
 
   class ZkBarrierListenerImpl implements ZkBarrierListener {
     private final String barrierAction = "BarrierAction";
+
     private long startTime = 0;
 
     @Override
@@ -301,10 +307,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
       metrics.barrierStateChange.inc();
       metrics.singleBarrierRebalancingTime.update(System.nanoTime() - startTime);
       if (ZkBarrierForVersionUpgrade.State.DONE.equals(state)) {
-        debounceTimer.scheduleAfterDebounceTime(
-            barrierAction,
-          0,
-          () -> onNewJobModelConfirmed(version));
+        debounceTimer.scheduleAfterDebounceTime(barrierAction, 0, () -> onNewJobModelConfirmed(version));
       } else {
         if (ZkBarrierForVersionUpgrade.State.TIMED_OUT.equals(state)) {
           // no-op
@@ -323,4 +326,50 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
       stop();
     }
   }
+
+  /// listener to handle session expiration
+  class ZkSessionStateChangedListener implements IZkStateListener {
+
+    @Override
+    public void handleStateChanged(Watcher.Event.KeeperState state)
+        throws Exception {
+      if (state == Watcher.Event.KeeperState.Expired) {
+        // if the session has expired it means that all the registration's ephemeral nodes are gone.
+        LOG.warn("Got session expired event for processor=" + processorId);
+
+        // increase generation of the ZK connection. All the callbacks from the previous generation will be ignored.
+        zkUtils.incGeneration();
+
+        if (coordinatorListener != null) {
+          coordinatorListener.onJobModelExpired();
+        }
+        // reset all the values that might have been from the previous session (e.g ephemeral node path)
+        zkUtils.unregister();
+
+      }
+    }
+
+    @Override
+    public void handleNewSession()
+        throws Exception {
+      LOG.info("Got new session created event for processor=" + processorId);
+
+
+      LOG.info("register zk controller for the new session");
+      zkController.register();
+    }
+
+    @Override
+    public void handleSessionEstablishmentError(Throwable error)
+        throws Exception {
+      // this means we cannot connect to zookeeper
+      LOG.info("handleSessionEstablishmentError received for processor=" + processorId, error);
+      stop();
+    }
+  }
+
+  @VisibleForTesting
+  public ZkUtils getZkUtils() {
+    return zkUtils;
+  }
 }
index 8caa5c6..97430cb 100644 (file)
@@ -61,7 +61,7 @@ public class ZkLeaderElector implements LeaderElector {
     this.zkUtils = zkUtils;
     this.keyBuilder = zkUtils.getKeyBuilder();
     this.hostName = getHostName();
-    this.previousProcessorChangeListener = new PreviousProcessorChangeListener();
+    this.previousProcessorChangeListener = new PreviousProcessorChangeListener(zkUtils);
 
     zkUtils.makeSurePersistentPathsExists(new String[]{keyBuilder.getProcessorsPath()});
   }
@@ -170,16 +170,25 @@ public class ZkLeaderElector implements LeaderElector {
   }
 
   // Only by non-leaders
-  class PreviousProcessorChangeListener implements IZkDataListener {
+  class PreviousProcessorChangeListener extends ZkUtils.GenIZkDataListener {
+
+    public PreviousProcessorChangeListener(ZkUtils zkUtils) {
+      super(zkUtils, "PreviousProcessorChangeListener");
+    }
     @Override
     public void handleDataChange(String dataPath, Object data) throws Exception {
       LOG.debug("Data change on path: " + dataPath + " Data: " + data);
+      if (notAValidEvent())
+        return;
     }
 
     @Override
-    public void handleDataDeleted(String dataPath) throws Exception {
-      LOG.info(
-          zLog("Data deleted on path " + dataPath + ". Predecessor went away. So, trying to become leader again..."));
+    public void handleDataDeleted(String dataPath)
+        throws Exception {
+      LOG.info(zLog("Data deleted on path " + dataPath + ". Predecessor went away. So, trying to become leader again..."));
+      if (notAValidEvent()) {
+        return;
+      }
       tryBecomeLeader();
     }
   }
index 7406cf5..8b6bc52 100644 (file)
@@ -26,6 +26,7 @@ import java.util.List;
 import java.util.Objects;
 import java.util.TreeSet;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import org.I0Itec.zkclient.IZkChildListener;
 import org.I0Itec.zkclient.IZkDataListener;
@@ -54,6 +55,14 @@ import org.slf4j.LoggerFactory;
  * </p>
  *
  * <p>
+ *  <b>Note on Session disconnect handling:</b>
+ *  After the session has timed out, and restored we may still get some notifications from before (from the old
+ *  session). To avoid this, we add a currentGeneration member, which starts with 0, and is increased each time
+ *  a new session is established. Current value of this member is passed to each Listener when it is created.
+ *  So if the Callback from this Listener comes with an old generation id - we ignore it.
+ * </p>
+ *
+ * <p>
  *   <b>Note on Session Management:</b>
  *   Session management, if needed, should be handled by the caller. This can be done by implementing
  *   {@link org.I0Itec.zkclient.IZkStateListener} and subscribing this listener to the current ZkClient. Note: The connection state change
@@ -68,24 +77,43 @@ public class ZkUtils {
   private volatile String ephemeralPath = null;
   private final ZkKeyBuilder keyBuilder;
   private final int connectionTimeoutMs;
+  private final AtomicInteger currentGeneration;
   private final ZkUtilsMetrics metrics;
 
+  public void incGeneration() {
+    currentGeneration.incrementAndGet();
+  }
+
+  public int getGeneration() {
+    return currentGeneration.get();
+  }
+
+
+
   public ZkUtils(ZkKeyBuilder zkKeyBuilder, ZkClient zkClient, int connectionTimeoutMs, MetricsRegistry metricsRegistry) {
     this.keyBuilder = zkKeyBuilder;
     this.connectionTimeoutMs = connectionTimeoutMs;
     this.zkClient = zkClient;
     this.metrics = new ZkUtilsMetrics(metricsRegistry);
+    this.currentGeneration = new AtomicInteger(0);
   }
 
   public void connect() throws ZkInterruptedException {
     boolean isConnected = zkClient.waitUntilConnected(connectionTimeoutMs, TimeUnit.MILLISECONDS);
     if (!isConnected) {
-      metrics.zkConnectionError.inc();
+      if (metrics != null) {
+        metrics.zkConnectionError.inc();
+      }
       throw new RuntimeException("Unable to connect to Zookeeper within connectionTimeout " + connectionTimeoutMs + "ms. Shutting down!");
     }
   }
 
-  ZkClient getZkClient() {
+  // reset all zk-session specific state
+  public void unregister() {
+    ephemeralPath = null;
+  }
+
+  public ZkClient getZkClient() {
     return zkClient;
   }
 
@@ -192,7 +220,9 @@ public class ZkUtils {
   String readProcessorData(String fullPath) {
     try {
       String data = zkClient.readData(fullPath, false);
-      metrics.reads.inc();
+      if (metrics != null) {
+        metrics.reads.inc();
+      }
       return data;
     } catch (Exception e) {
       throw new SamzaException(String.format("Cannot read ZK node: %s", fullPath), e);
@@ -216,12 +246,11 @@ public class ZkUtils {
     String processorPath = keyBuilder.getProcessorsPath();
     List<String> processorIds = new ArrayList<>(znodeIds.size());
     if (znodeIds.size() > 0) {
-
       for (String child : znodeIds) {
         String fullPath = String.format("%s/%s", processorPath, child);
         processorIds.add(new ProcessorData(readProcessorData(fullPath)).getProcessorId());
       }
-
+      Collections.sort(processorIds);
       LOG.info("Found these children - " + znodeIds);
       LOG.info("Found these processorIds - " + processorIds);
     }
@@ -235,12 +264,16 @@ public class ZkUtils {
 
   public void subscribeDataChanges(String path, IZkDataListener dataListener) {
     zkClient.subscribeDataChanges(path, dataListener);
-    metrics.subscriptions.inc();
+    if (metrics != null) {
+      metrics.subscriptions.inc();
+    }
   }
 
   public void subscribeChildChanges(String path, IZkChildListener listener) {
     zkClient.subscribeChildChanges(path, listener);
-    metrics.subscriptions.inc();
+    if (metrics != null) {
+      metrics.subscriptions.inc();
+    }
   }
 
   public void unsubscribeChildChanges(String path, IZkChildListener childListener) {
@@ -249,7 +282,9 @@ public class ZkUtils {
 
   public void writeData(String path, Object object) {
     zkClient.writeData(path, object);
-    metrics.writes.inc();
+    if (metrics != null) {
+      metrics.writes.inc();
+    }
   }
 
   public boolean exists(String path) {
@@ -261,13 +296,65 @@ public class ZkUtils {
   }
 
   /**
+   * Generation enforcing zk listener abstract class.
+   * It helps listeners, which extend it, to notAValidEvent old generation events.
+   * We cannot use 'sessionId' for this because it is not available through ZkClient (at leaste without reflection)
+   */
+  public abstract static class GenIZkChildListener implements IZkChildListener {
+    private final int generation;
+    private final ZkUtils zkUtils;
+    private final String listenerName;
+
+    public GenIZkChildListener(ZkUtils zkUtils, String listenerName) {
+      generation = zkUtils.getGeneration();
+      this.zkUtils = zkUtils;
+      this.listenerName = listenerName;
+    }
+
+    protected boolean notAValidEvent() {
+      int curGeneration = zkUtils.getGeneration();
+      if (curGeneration != generation) {
+        LOG.warn("SKIPPING handleDataChanged for " + listenerName +
+            " from wrong generation. current generation=" + curGeneration + "; callback generation= " + generation);
+        return true;
+      }
+      return false;
+    }
+  }
+
+  public abstract static class GenIZkDataListener implements IZkDataListener {
+    private final int generation;
+    private final ZkUtils zkUtils;
+    private final String listenerName;
+
+    public GenIZkDataListener(ZkUtils zkUtils, String listenerName) {
+      generation = zkUtils.getGeneration();
+      this.zkUtils = zkUtils;
+      this.listenerName = listenerName;
+    }
+
+    protected boolean notAValidEvent() {
+      int curGeneration = zkUtils.getGeneration();
+      if (curGeneration != generation) {
+        LOG.warn("SKIPPING handleDataChanged for " + listenerName +
+            " from wrong generation. curGen=" + curGeneration + "; cb gen= " + generation);
+        return true;
+      }
+      return false;
+    }
+
+  }
+
+  /**
     * subscribe for changes of JobModel version
     * @param dataListener describe this
     */
-  public void subscribeToJobModelVersionChange(IZkDataListener dataListener) {
+  public void subscribeToJobModelVersionChange(GenIZkDataListener dataListener) {
     LOG.info(" subscribing for jm version change at:" + keyBuilder.getJobModelVersionPath());
     zkClient.subscribeDataChanges(keyBuilder.getJobModelVersionPath(), dataListener);
-    metrics.subscriptions.inc();
+    if (metrics != null) {
+      metrics.subscriptions.inc();
+    }
   }
 
   /**
@@ -345,11 +432,10 @@ public class ZkUtils {
       LOG.error(msg, e);
       throw new SamzaException(msg, e);
     }
-    LOG.info("published new version: " + newVersion + "; expected data version = " + (dataVersion  + 1) +
-        "(actual data version after update = " + stat.getVersion() +    ")");
+    LOG.info("published new version: " + newVersion + "; expected data version = " + (dataVersion + 1) +
+        "(actual data version after update = " + stat.getVersion() + ")");
   }
 
-
   /**
    * verify that given paths exist in ZK
    * @param paths - paths to verify or create
index 2545194..1b3b893 100644 (file)
@@ -48,7 +48,7 @@ object JobConfig {
   val JOB_CONTAINER_SINGLE_THREAD_MODE = "job.container.single.thread.mode"
   val JOB_INTERMEDIATE_STREAM_PARTITIONS = "job.intermediate.stream.partitions"
   val JOB_DEBOUNCE_TIME_MS = "job.debounce.time.ms"
-  val DEFAULT_DEBOUNCE_TIME_MS = 2000
+  val DEFAULT_DEBOUNCE_TIME_MS = 20000
 
   val SSP_GROUPER_FACTORY = "job.systemstreampartition.grouper.factory"
 
index 3bf5c95..b830579 100644 (file)
@@ -32,7 +32,7 @@ import org.apache.samza.checkpoint.{CheckpointListener, CheckpointManagerFactory
 import org.apache.samza.config.JobConfig.Config2Job
 import org.apache.samza.config.MetricsConfig.Config2Metrics
 import org.apache.samza.config.SerializerConfig.Config2Serializer
-import org.apache.samza.config.{ClusterManagerConfig, Config, ShellCommandConfig, StorageConfig}
+import org.apache.samza.config._
 import org.apache.samza.config.StorageConfig.Config2Storage
 import org.apache.samza.config.StreamConfig.Config2Stream
 import org.apache.samza.config.SystemConfig.Config2System
@@ -649,7 +649,7 @@ class SamzaContainer(
   jvm: JvmMetrics = null,
   taskThreadPool: ExecutorService = null) extends Runnable with Logging {
 
-  val shutdownMs = containerContext.config.getShutdownMs.getOrElse(5000L)
+  val shutdownMs = containerContext.config.getShutdownMs.getOrElse(TaskConfigJava.DEFAULT_TASK_SHUTDOWN_MS)
   var shutdownHookThread: Thread = null
   var jmxServer: JmxServer = null
 
index 6f0b53a..243716e 100644 (file)
@@ -80,7 +80,7 @@ class SystemConsumers (
 
   /**
    * This parameter is to define how to deal with deserialization failure. If
-   * set to true, the task will skip the messages when deserialization fails.
+   * set to true, the task will notAValidEvent the messages when deserialization fails.
    * If set to false, the task will throw SamzaException and fail the container.
    */
   dropDeserializationError: Boolean = SystemConsumers.DEFAULT_DROP_SERIALIZATION_ERROR,
index 5682a7c..14c7fd1 100644 (file)
@@ -88,7 +88,7 @@ public class TestMultiFileHdfsReader {
     // read until the middle of the second file
     multiReader = new MultiFileHdfsReader(HdfsReaderFactory.ReaderType.AVRO, ssp,
       Arrays.asList(descriptors), offset);
-    multiReader.readNext(); // skip one duplicate event
+    multiReader.readNext(); // notAValidEvent one duplicate event
     for (; index < NUM_EVENTS + NUM_EVENTS / 2; index++) {
       IncomingMessageEnvelope envelope = multiReader.readNext();
       GenericRecord record = (GenericRecord) envelope.getMessage();
@@ -101,7 +101,7 @@ public class TestMultiFileHdfsReader {
     // read the rest of all files
     multiReader = new MultiFileHdfsReader(HdfsReaderFactory.ReaderType.AVRO, ssp,
       Arrays.asList(descriptors), offset);
-    multiReader.readNext(); // skip one duplicate event
+    multiReader.readNext(); // notAValidEvent one duplicate event
     while (multiReader.hasNext()) {
       IncomingMessageEnvelope envelope = multiReader.readNext();
       GenericRecord record = (GenericRecord) envelope.getMessage();
@@ -116,7 +116,7 @@ public class TestMultiFileHdfsReader {
     // reopen with the offset of the last record
     multiReader = new MultiFileHdfsReader(HdfsReaderFactory.ReaderType.AVRO, ssp,
       Arrays.asList(descriptors), offset);
-    multiReader.readNext(); // skip one duplicate event
+    multiReader.readNext(); // notAValidEvent one duplicate event
     Assert.assertFalse(multiReader.hasNext());
     multiReader.close();
   }
index aa13fd8..e033b53 100644 (file)
@@ -202,7 +202,7 @@ private[kafka] class KafkaSystemConsumer(
           // we need to lock.
           this.synchronized {
             // Check if we still need this TopicAndPartition inside the
-            // critical section. If we don't, then skip it.
+            // critical section. If we don't, then notAValidEvent it.
             topicPartitionsAndOffsets.get(head) match {
               case Some(nextOffset) =>
                 getHostPort(topicMetadata(head.topic), head.partition) match {
index 2aac6aa..8385c4a 100644 (file)
@@ -96,7 +96,7 @@ class KeyValueStorageEngine[K, V](
 
   /**
    * Restore the contents of this key/value store from the change log,
-   * batching updates to underlying raw store to skip wrapping functions for efficiency.
+   * batching updates to underlying raw store to notAValidEvent wrapping functions for efficiency.
    */
   def restore(envelopes: java.util.Iterator[IncomingMessageEnvelope]) {
     val batch = new java.util.ArrayList[Entry[Array[Byte], Array[Byte]]](batchSize)
  * under the License.
  */
 
-package org.apache.samza.test.processor;
+package org.apache.samza.processor;
 
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import org.apache.samza.processor.StreamProcessor;
-import org.apache.samza.processor.TestZkStreamProcessorBase;
 import org.apache.samza.zk.TestZkUtils;
 import org.junit.Assert;
 import org.junit.Test;
@@ -63,42 +60,37 @@ public class TestZkStreamProcessor extends TestZkStreamProcessorBase {
     // initialize the processors
     StreamProcessor[] streamProcessors = new StreamProcessor[processorIds.length];
     // we need to know when the processor has started
-    Object[] startWait = new Object[processorIds.length];
+    CountDownLatch[] startWait = new CountDownLatch[processorIds.length];
     for (int i = 0; i < processorIds.length; i++) {
-      startWait[i] = new Object();
+      startWait[i] = new CountDownLatch(1);
       streamProcessors[i] = createStreamProcessor(processorIds[i], map, startWait[i], null);
     }
 
-    // produce messageCount messages, starting with key 0
-    produceMessages(0, inputTopic, messageCount);
 
     // run the processors in separate threads
     Thread[] threads = new Thread[processorIds.length];
+    CountDownLatch[] stopLatches = new CountDownLatch[processorIds.length];
     for (int i = 0; i < processorIds.length; i++) {
-      threads[i] = runInThread(streamProcessors[i], TestZkStreamProcessorBase.TestStreamTask.endLatch);
+      stopLatches[i] = new CountDownLatch(1);
+      threads[i] = runInThread(streamProcessors[i], stopLatches[i]);
       threads[i].start();
-      // wait until the processor reports that it has started
-      try {
-        synchronized (startWait[i]) {
-          startWait[i].wait(1000);
-        }
-      } catch (InterruptedException e) {
-        Assert.fail("got interrupted while waiting for the " + i + "th processor to start.");
-      }
     }
+    // wait until the processor reports that it has started
+    for (int i = 0; i < processorIds.length; i++) {
+      waitForProcessorToStartStop(startWait[i]);
+    }
+
+    // produce messageCount messages, starting with key 0
+    produceMessages(0, inputTopic, messageCount);
 
     // wait until all the events are consumed
-    try {
-      TestZkStreamProcessorBase.TestStreamTask.endLatch.await(10, TimeUnit.SECONDS);
-    } catch (InterruptedException e) {
-      Assert.fail("endLatch.await failed with an interruption:" + e.getLocalizedMessage());
-    }
+    waitUntilMessagesLeftN(0);
 
     // collect all the threads
     try {
-      for (Thread t : threads) {
-        stopProcessor(t);
-        t.join(1000);
+      for (int i = 0; i < threads.length; i++) {
+        stopProcessor(stopLatches[i]);
+        threads[i].join(1000);
       }
     } catch (InterruptedException e) {
       Assert.fail("Failed to join finished thread:" + e.getLocalizedMessage());
@@ -120,15 +112,16 @@ public class TestZkStreamProcessor extends TestZkStreamProcessorBase {
     TestStreamTask.endLatch = new CountDownLatch(totalEventsToGenerate);
 
     // create first processor
-    Object startWait1 = new Object();
-    Object stopWait1 = new Object();
-    StreamProcessor sp = createStreamProcessor("20", map, startWait1, stopWait1);
+    CountDownLatch startWait1 = new CountDownLatch(1);
+    CountDownLatch stopWait1 = new CountDownLatch(1);
+    StreamProcessor sp1 = createStreamProcessor("20", map, startWait1, stopWait1);
 
     // produce first batch of messages starting with 0
     produceMessages(0, inputTopic, messageCount);
 
     // start the first processor
-    Thread t1 = runInThread(sp, TestStreamTask.endLatch);
+    CountDownLatch stopLatch1 = new CountDownLatch(1);
+    Thread t1 = runInThread(sp1, stopLatch1);
     t1.start();
 
     // wait until the processor reports that it has started
@@ -136,18 +129,22 @@ public class TestZkStreamProcessor extends TestZkStreamProcessorBase {
 
     // make sure it consumes all the messages from the first batch
     waitUntilMessagesLeftN(totalEventsToGenerate - messageCount);
+    CountDownLatch containerStopped1 = sp1.jcContainerShutdownLatch;
 
     // start the second processor
-    Object startWait2 = new Object();
+    CountDownLatch startWait2 = new CountDownLatch(1);
     StreamProcessor sp2 = createStreamProcessor("21", map, startWait2, null);
-    Thread t2 = runInThread(sp2, TestStreamTask.endLatch);
+
+    CountDownLatch stopLatch2 = new CountDownLatch(1);
+    Thread t2 = runInThread(sp2, stopLatch2);
     t2.start();
 
     // wait until 2nd processor reports that it has started
     waitForProcessorToStartStop(startWait2);
 
-    // wait until the 1st processor reports that it has stopped
-    waitForProcessorToStartStop(stopWait1);
+    // wait until the 1st processor reports that it has stopped its container
+    LOG.info("containerStopped latch = " + containerStopped1);
+    waitForProcessorToStartStop(containerStopped1);
 
     // let the system to publish and distribute the new job model
     TestZkUtils.sleepMs(600);
@@ -160,8 +157,8 @@ public class TestZkStreamProcessor extends TestZkStreamProcessorBase {
 
     // shutdown both
     try {
-      stopProcessor(t1);
-      stopProcessor(t2);
+      stopProcessor(stopLatch1);
+      stopProcessor(stopLatch2);
       t1.join(1000);
       t2.join(1000);
     } catch (InterruptedException e) {
@@ -186,19 +183,22 @@ public class TestZkStreamProcessor extends TestZkStreamProcessorBase {
     TestStreamTask.endLatch = new CountDownLatch(totalEventsToGenerate);
 
     // create first processor
-    Object waitStart1 = new Object();
-    Object waitStop1 = new Object();
+    CountDownLatch waitStart1 = new CountDownLatch(1);
+    CountDownLatch waitStop1 = new CountDownLatch(1);
     StreamProcessor sp1 = createStreamProcessor("30", map, waitStart1, waitStop1);
 
     // start the first processor
-    Thread t1 = runInThread(sp1, TestStreamTask.endLatch);
+    CountDownLatch stopLatch1 = new CountDownLatch(1);
+    Thread t1 = runInThread(sp1, stopLatch1);
     t1.start();
 
     // start the second processor
-    Object waitStart2 = new Object();
-    Object waitStop2 = new Object();
+    CountDownLatch waitStart2 = new CountDownLatch(1);
+    CountDownLatch waitStop2 = new CountDownLatch(1);
     StreamProcessor sp2 = createStreamProcessor("31", map, waitStart2, waitStop2);
-    Thread t2 = runInThread(sp2, TestStreamTask.endLatch);
+
+    CountDownLatch stopLatch2 = new CountDownLatch(1);
+    Thread t2 = runInThread(sp2, stopLatch2);
     t2.start();
 
     // wait until the processor reports that it has started
@@ -212,15 +212,19 @@ public class TestZkStreamProcessor extends TestZkStreamProcessorBase {
 
     // make sure they consume all the messages from the first batch
     waitUntilMessagesLeftN(totalEventsToGenerate - messageCount);
+    CountDownLatch containerStopped2 = sp2.jcContainerShutdownLatch;
 
     // stop the first processor
-    stopProcessor(t1);
+    stopProcessor(stopLatch1);
 
     // wait until it's really down
     waitForProcessorToStartStop(waitStop1);
 
-    // processor1 will stop and start again. We wait for its stop to make sure we can count EXACTLY how many messages it reads.
-    waitForProcessorToStartStop(waitStop2);
+    // processor2 will stop it container and start again.
+    // We wait for the container's stop to make sure we can count EXACTLY how many messages it reads.
+
+    LOG.info("containerStopped latch = " + containerStopped2);
+    waitForProcessorToStartStop(containerStopped2);
 
     // let the system to publish and distribute the new job model
     TestZkUtils.sleepMs(300);
@@ -234,7 +238,7 @@ public class TestZkStreamProcessor extends TestZkStreamProcessorBase {
     // shutdown p2
 
     try {
-      stopProcessor(t2);
+      stopProcessor(stopLatch2);
       t2.join(1000);
     } catch (InterruptedException e) {
       Assert.fail("Failed to join finished thread:" + e.getLocalizedMessage());
index a315083..4cbe252 100644 (file)
@@ -19,6 +19,7 @@
 
 package org.apache.samza.processor;
 
+import java.lang.reflect.Field;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -26,8 +27,11 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import kafka.utils.TestUtils;
+import org.I0Itec.zkclient.ZkClient;
+import org.I0Itec.zkclient.ZkConnection;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -53,6 +57,8 @@ import org.apache.samza.test.StandaloneIntegrationTestHarness;
 import org.apache.samza.test.StandaloneTestUtils;
 import org.apache.samza.util.Util;
 import org.apache.samza.zk.TestZkUtils;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.server.ZooKeeperServer;
 import org.junit.Assert;
 import org.junit.Before;
 import org.slf4j.Logger;
@@ -93,8 +99,28 @@ public class TestZkStreamProcessorBase extends StandaloneIntegrationTestHarness
     createTopics(inputTopic, outputTopic);
   }
 
-  protected StreamProcessor createStreamProcessor(final String pId, Map<String, String> map, final Object mutexStart,
-      final Object mutexStop) {
+  // session expiration simulation
+  // have to use the reflection to get the session id
+  protected void expireSession(ZkClient zkClient) {
+    ZkConnection zkConnection = null;
+    try {
+      Field privateField = ZkClient.class.getDeclaredField("_connection");
+      privateField.setAccessible(true);
+      zkConnection = (ZkConnection) privateField.get(zkClient);
+    } catch (NoSuchFieldException | IllegalAccessException e) {
+      Assert.fail(e.toString());
+    }
+
+    ZooKeeper zookeeper = zkConnection.getZookeeper();
+    long sessionId = zookeeper.getSessionId();
+
+    LOG.info("Closing/expiring session:" + sessionId);
+    ZooKeeperServer zkServer = zookeeper().zookeeper();
+    zkServer.closeSession(sessionId);
+  }
+
+  protected StreamProcessor createStreamProcessor(final String pId, Map<String, String> map, final CountDownLatch waitStart,
+      final CountDownLatch waitStop) {
     map.put(ApplicationConfig.PROCESSOR_ID, pId);
 
     Config config = new MapConfig(map);
@@ -105,20 +131,16 @@ public class TestZkStreamProcessorBase extends StandaloneIntegrationTestHarness
     StreamProcessorLifecycleListener listener = new StreamProcessorLifecycleListener() {
       @Override
       public void onStart() {
-        if (mutexStart != null) {
-          synchronized (mutexStart) {
-            mutexStart.notifyAll();
-          }
+        if (waitStart != null) {
+            waitStart.countDown();
         }
         LOG.info("onStart is called for pid=" + pId);
       }
 
       @Override
       public void onShutdown() {
-        if (mutexStop != null) {
-          synchronized (mutexStart) {
-            mutexStart.notify();
-          }
+        if (waitStop != null) {
+          waitStop.countDown();
         }
         LOG.info("onShutdown is called for pid=" + pId);
       }
@@ -144,7 +166,7 @@ public class TestZkStreamProcessorBase extends StandaloneIntegrationTestHarness
       int messageCount) {
     Map<String, String> configs = new HashMap<>();
     configs.putAll(StandaloneTestUtils
-        .getStandaloneConfigs("test-job", "org.apache.samza.test.processor.TestZkStreamProcessor.TestStreamTask"));
+        .getStandaloneConfigs("test-job", "org.apache.samza.processor.TestZkStreamProcessor.TestStreamTask"));
     configs.putAll(StandaloneTestUtils
         .getKafkaSystemConfigs(testSystem, bootstrapServers(), zkConnect(), null, StandaloneTestUtils.SerdeAlias.STRING,
             true));
@@ -182,23 +204,27 @@ public class TestZkStreamProcessorBase extends StandaloneIntegrationTestHarness
    * Runs the provided stream processor by starting it, waiting on the provided latch with a timeout,
    * and then stopping it.
    */
-  protected Thread runInThread(final StreamProcessor processor, CountDownLatch latch) {
+  protected Thread runInThread(final StreamProcessor processor, CountDownLatch stopStartLatch) {
     Thread t = new Thread() {
 
       @Override
       public void run() {
+        LOG.info("about to start processor " + processor);
         processor.start();
+        LOG.info("started processor " + processor);
         try {
           // just wait
-          synchronized (this) {
-            this.wait(100000);
+          if (!stopStartLatch.await(1000000, TimeUnit.MILLISECONDS)) {
+            LOG.warn("Wait timed out for processor " + processor);
+            Assert.fail("Wait timed out for processor " + processor);
           }
-          LOG.info("notified. Abandon the wait.");
+          LOG.info("notified. Abandon the wait for processor " + processor);
         } catch (InterruptedException e) {
           LOG.error("wait interrupted" + e);
         }
-        LOG.info("Stopping the processor");
+        LOG.info("Stopping the processor" + processor);
         processor.stop();
+        LOG.info("Stopped the processor" + processor);
       }
     };
     return t;
@@ -261,26 +287,25 @@ public class TestZkStreamProcessorBase extends StandaloneIntegrationTestHarness
         System.out.println("2read all. current count = " + leftEventsCount);
         break;
       }
-      TestZkUtils.sleepMs(1000);
+      TestZkUtils.sleepMs(3000);
       attempts--;
     }
     Assert.assertTrue("Didn't read all the leftover events in " + ATTEMPTS_NUMBER + " attempts", attempts > 0);
   }
 
-  protected void waitForProcessorToStartStop(Object waitObject) {
+  protected void waitForProcessorToStartStop(CountDownLatch waitObject) {
+    LOG.info("Waiting on " + waitObject);
     try {
-      synchronized (waitObject) {
-        waitObject.wait(1000);
+      if (!waitObject.await(30000, TimeUnit.MILLISECONDS)) {
+        Assert.fail("Timed out while waiting for the processor to start/stop.");
       }
     } catch (InterruptedException e) {
-      Assert.fail("got interrupted while waiting for the first processor to start.");
+      Assert.fail("Got interrupted while waiting for the processor to start/stop.");
     }
   }
 
-  protected void stopProcessor(Thread threadName) {
-    synchronized (threadName) {
-      threadName.notify();
-    }
+  protected void stopProcessor(CountDownLatch stopLatch) {
+    stopLatch.countDown();
   }
 
   // StreamTaskClass
  * under the License.
  */
 
-package org.apache.samza.test.processor;
+package org.apache.samza.processor;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.ZkConfig;
-import org.apache.samza.processor.StreamProcessor;
-import org.apache.samza.processor.TestZkStreamProcessorBase;
 import org.apache.samza.zk.TestZkUtils;
 import org.junit.Assert;
 import org.junit.Before;
@@ -82,18 +80,21 @@ public class TestZkStreamProcessorFailures extends TestZkStreamProcessorBase {
 
     TestStreamTask.endLatch = new CountDownLatch(totalEventsToBeConsumed);
     // create first processor
-    Object waitStart1 = new Object();
-    Object waitStop1 = new Object();
+    CountDownLatch waitStart1 = new CountDownLatch(1);
+    CountDownLatch waitStop1 = new CountDownLatch(1);
     StreamProcessor sp1 = createStreamProcessor("101", map, waitStart1, waitStop1);
     // start the first processor
-    Thread t1 = runInThread(sp1, TestStreamTask.endLatch);
+    CountDownLatch stopLatch1 = new CountDownLatch(1);
+    Thread t1 = runInThread(sp1, stopLatch1);
     t1.start();
 
     // start the second processor
-    Object waitStart2 = new Object();
-    Object waitStop2 = new Object();
+    CountDownLatch waitStart2 = new CountDownLatch(1);
+    CountDownLatch waitStop2 = new CountDownLatch(1);
     StreamProcessor sp2 = createStreamProcessor("102", map, waitStart2, waitStop2);
-    Thread t2 = runInThread(sp2, TestStreamTask.endLatch);
+
+    CountDownLatch stopLatch2 = new CountDownLatch(1);
+    Thread t2 = runInThread(sp2, stopLatch2);
     t2.start();
 
     // wait until the 1st processor reports that it has started
@@ -107,14 +108,17 @@ public class TestZkStreamProcessorFailures extends TestZkStreamProcessorBase {
 
     // make sure they consume all the messages
     waitUntilMessagesLeftN(totalEventsToBeConsumed - messageCount);
+    CountDownLatch containerStopped1 = sp1.jcContainerShutdownLatch;
+    CountDownLatch containerStopped2 = sp2.jcContainerShutdownLatch;
 
     // produce the bad messages
     produceMessages(BAD_MESSAGE_KEY, inputTopic, 4);
 
-    waitForProcessorToStartStop(waitStop1);
+    waitForProcessorToStartStop(
+        containerStopped1); // TODO: after container failure propagates to StreamProcessor change back
 
-    // wait until the 2nd processor reports that it has stopped
-    waitForProcessorToStartStop(waitStop2);
+    // wait until the 2nd processor reports that it has stopped its container
+    waitForProcessorToStartStop(containerStopped2);
 
     // give some extra time to let the system to publish and distribute the new job model
     TestZkUtils.sleepMs(300);
@@ -127,7 +131,7 @@ public class TestZkStreamProcessorFailures extends TestZkStreamProcessorBase {
 
     // shutdown p2
     try {
-      stopProcessor(t2);
+      stopProcessor(stopLatch2);
       t2.join(1000);
     } catch (InterruptedException e) {
       Assert.fail("Failed to join finished thread:" + e.getLocalizedMessage());
diff --git a/samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorSession.java b/samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorSession.java
new file mode 100644 (file)
index 0000000..10b08d9
--- /dev/null
@@ -0,0 +1,132 @@
+/*
+ * 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 org.apache.samza.processor;
+
+import java.util.concurrent.CountDownLatch;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.ZkConfig;
+import org.apache.samza.zk.ZkJobCoordinator;
+import org.junit.Assert;
+import org.junit.Test;
+
+
+/**
+ * Happy path tests.
+ * Start 1, 2, 5 processors and make sure they all consume all the events.
+ */
+public class TestZkStreamProcessorSession extends TestZkStreamProcessorBase {
+
+  @Override
+  protected String prefix() {
+    return "test_ZKS_";
+  }
+
+  @Test
+  public void testSingleStreamProcessor() {
+    testStreamProcessorWithSessionRestart(new String[]{"1"});
+  }
+
+  @Test
+  public void testTwoStreamProcessors() {
+    testStreamProcessorWithSessionRestart(new String[]{"2", "3"});
+  }
+
+  @Test
+  public void testFiveStreamProcessors() {
+    testStreamProcessorWithSessionRestart(new String[]{"4", "5", "6", "7", "8"});
+  }
+
+  private void testStreamProcessorWithSessionRestart(String[] processorIds) {
+
+    // set shorter session expiration for the test
+    map.put(ZkConfig.ZK_SESSION_TIMEOUT_MS, "500");
+    map.put(JobConfig.JOB_DEBOUNCE_TIME_MS(), "3000");
+
+    // create a latch of the size equals to the number of messages
+    int totalEventsToGenerate = 3 * messageCount;
+    TestZkStreamProcessorBase.TestStreamTask.endLatch = new CountDownLatch(totalEventsToGenerate);
+
+    // initialize the processors
+    StreamProcessor[] streamProcessors = new StreamProcessor[processorIds.length];
+    ZkJobCoordinator[] jobCoordinators = new ZkJobCoordinator[processorIds.length];
+    Thread[] threads = new Thread[processorIds.length];
+    CountDownLatch[] threadStopLatches = new CountDownLatch[processorIds.length];
+    CountDownLatch[] containerStopLatches = new CountDownLatch[processorIds.length];
+    // we need to know when the processor has started
+    CountDownLatch[] startWait = new CountDownLatch[processorIds.length];
+    //CountDownLatch[] stopWait = new CountDownLatch[processorIds.length];
+
+    for (int i = 0; i < processorIds.length; i++) {
+      startWait[i] = new CountDownLatch(1);
+      streamProcessors[i] = createStreamProcessor(processorIds[i], map, startWait[i], null);
+      jobCoordinators[i] = (ZkJobCoordinator) streamProcessors[i].getCurrentJobCoordinator();
+    }
+
+    // run the processors in separate threads
+    for (int i = 0; i < processorIds.length; i++) {
+      threadStopLatches[i] = new CountDownLatch(1); // is used int stopProcessor
+      threads[i] = runInThread(streamProcessors[i], threadStopLatches[i]);
+      threads[i].start();
+    }
+
+    for (int i = 0; i < processorIds.length; i++) {
+      // wait until the processor reports that it has started
+      waitForProcessorToStartStop(startWait[i]);
+    }
+
+    // produce messageCount messages, starting with key 0
+    produceMessages(0, inputTopic, messageCount);
+
+    // make sure it consumes all the messages from the first batch
+    waitUntilMessagesLeftN(totalEventsToGenerate - messageCount);
+
+    // Get the container stop latch to be able to check when a container is stopped.
+    // New jcContainerShutdownLatch is created after each onNewJobModel,
+    // so we need to get the current one, before it changed..
+    for (int i = 0; i < processorIds.length; i++) {
+      containerStopLatches[i] = streamProcessors[i].jcContainerShutdownLatch;
+    }
+
+    // expire zk session of one of the processors
+    expireSession(jobCoordinators[0].getZkUtils().getZkClient());
+
+    // wait until all other processors report that they have stopped their containers
+    for (int i = 0; i < processorIds.length; i++) {
+      waitForProcessorToStartStop(containerStopLatches[i]);
+    }
+
+    produceMessages(messageCount, inputTopic, messageCount);
+
+    waitUntilMessagesLeftN(0);
+
+    // collect all the threads
+    try {
+      for (int i = 0; i < threads.length; i++) {
+        stopProcessor(threadStopLatches[i]);
+        threads[i].join(1000);
+      }
+    } catch (InterruptedException e) {
+      Assert.fail("Failed to join finished thread:" + e.getLocalizedMessage());
+    }
+
+    verifyNumMessages(outputTopic, 2 * messageCount, totalEventsToGenerate);
+  }
+}
+