From c86ec0a22b8e18f544343609a3d8b423cd85658a Mon Sep 17 00:00:00 2001
From: Bharath Kumarasubramanian <bkumaras@linkedin.com>
Date: Fri, 26 May 2017 13:14:17 -0700
Subject: [PATCH 1/7] Extract hdfs docs into its own section

---
 docs/learn/documentation/versioned/index.html | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/docs/learn/documentation/versioned/index.html b/docs/learn/documentation/versioned/index.html
index a710383ebb..95468907c7 100644
--- a/docs/learn/documentation/versioned/index.html
+++ b/docs/learn/documentation/versioned/index.html
@@ -75,6 +75,13 @@ <h4>Job Deployment</h4>
   <li><a href="jobs/split-deployment.html">Separating Samza Framework and Jobs Deployment</a></li>
 </ul>
 
+<h4>Batch Processing</h4>
+
+<ul class="documentation-list">
+  <li><a href="hdfs/consumer.html">Reading from HDFS</a></li>
+  <li><a href="hdfs/producer.html">Writing to HDFS</a></li>
+</ul>
+
 <h4>YARN</h4>
 
 <ul class="documentation-list">
@@ -83,8 +90,6 @@ <h4>YARN</h4>
   <li><a href="yarn/yarn-host-affinity.html">Host Affinity & Yarn</a></li>
   <li><a href="yarn/yarn-resource-localization.html">Resource Localization</a></li>
   <li><a href="yarn/yarn-security.html">Yarn Security</a></li>
-  <li><a href="hdfs/producer.html">Writing to HDFS</a></li>
-  <li><a href="hdfs/consumer.html">Reading from HDFS</a></li>
 <!-- TODO write yarn pages
   <li><a href="">Fault Tolerance</a></li>
   <li><a href="">Security</a></li>

From e2ed4c8fe5d2976d0783e8b3fe978dc7ebcd4483 Mon Sep 17 00:00:00 2001
From: Bharath Kumarasubramanian <bkumaras@linkedin.com>
Date: Wed, 9 Aug 2017 15:11:25 -0700
Subject: [PATCH 2/7] SAMZA-1385: Fix zookeeper path conflict between
 LocalApplicationRunner and ZkJobCoordinator

---
 docs/learn/documentation/versioned/index.html |  9 ++-----
 .../samza/runtime/LocalApplicationRunner.java | 27 ++++++++++---------
 2 files changed, 17 insertions(+), 19 deletions(-)

diff --git a/docs/learn/documentation/versioned/index.html b/docs/learn/documentation/versioned/index.html
index 95468907c7..a710383ebb 100644
--- a/docs/learn/documentation/versioned/index.html
+++ b/docs/learn/documentation/versioned/index.html
@@ -75,13 +75,6 @@ <h4>Job Deployment</h4>
   <li><a href="jobs/split-deployment.html">Separating Samza Framework and Jobs Deployment</a></li>
 </ul>
 
-<h4>Batch Processing</h4>
-
-<ul class="documentation-list">
-  <li><a href="hdfs/consumer.html">Reading from HDFS</a></li>
-  <li><a href="hdfs/producer.html">Writing to HDFS</a></li>
-</ul>
-
 <h4>YARN</h4>
 
 <ul class="documentation-list">
@@ -90,6 +83,8 @@ <h4>YARN</h4>
   <li><a href="yarn/yarn-host-affinity.html">Host Affinity & Yarn</a></li>
   <li><a href="yarn/yarn-resource-localization.html">Resource Localization</a></li>
   <li><a href="yarn/yarn-security.html">Yarn Security</a></li>
+  <li><a href="hdfs/producer.html">Writing to HDFS</a></li>
+  <li><a href="hdfs/consumer.html">Reading from HDFS</a></li>
 <!-- TODO write yarn pages
   <li><a href="">Fault Tolerance</a></li>
   <li><a href="">Security</a></li>
diff --git a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
index b0bfc8a62d..20e26de66c 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
@@ -26,6 +26,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.samza.SamzaException;
@@ -59,11 +60,11 @@ public class LocalApplicationRunner extends AbstractApplicationRunner {
   private static final Logger LOG = LoggerFactory.getLogger(LocalApplicationRunner.class);
   // Latch id that's used for awaiting the init of application before creating the StreamProcessors
   private static final String INIT_LATCH_ID = "init";
+  private static final String APPLICATION_RUNNER_ZK_PATH_SUFFIX = "/" + "ApplicationRunnerData";
   // Latch timeout is set to 10 min
   private static final long LATCH_TIMEOUT_MINUTES = 10;
 
   private final String uid;
-  private final CoordinationUtils coordinationUtils;
   private final Set<StreamProcessor> processors = ConcurrentHashMap.newKeySet();
   private final CountDownLatch shutdownLatch = new CountDownLatch(1);
   private final AtomicInteger numProcessorsToStart = new AtomicInteger();
@@ -122,9 +123,6 @@ private void shutdownAndNotify() {
         }
       }
 
-      if (coordinationUtils != null) {
-        coordinationUtils.reset();
-      }
       shutdownLatch.countDown();
     }
   }
@@ -132,7 +130,6 @@ private void shutdownAndNotify() {
   public LocalApplicationRunner(Config config) {
     super(config);
     uid = UUID.randomUUID().toString();
-    coordinationUtils = createCoordinationUtils();
   }
 
   @Override
@@ -216,7 +213,8 @@ public void waitForFinish() {
     // TODO: we will need a better way to package the configs with application runner
     if (ZkJobCoordinatorFactory.class.getName().equals(jobCoordinatorFactoryClassName)) {
       ApplicationConfig appConfig = new ApplicationConfig(config);
-      return new ZkCoordinationServiceFactory().getCoordinationService(appConfig.getGlobalAppId(), uid, config);
+      return new ZkCoordinationServiceFactory().getCoordinationService(
+          appConfig.getGlobalAppId() + APPLICATION_RUNNER_ZK_PATH_SUFFIX, uid, config);
     } else {
       return null;
     }
@@ -230,17 +228,22 @@ public void waitForFinish() {
    * @param intStreams list of intermediate {@link StreamSpec}s
    * @throws Exception exception for latch timeout
    */
-  /* package private */ void createStreams(List<StreamSpec> intStreams) throws Exception {
+  /* package private */ void createStreams(List<StreamSpec> intStreams) throws TimeoutException {
     if (!intStreams.isEmpty()) {
+      CoordinationUtils coordinationUtils = createCoordinationUtils();
       if (coordinationUtils != null) {
-        Latch initLatch = coordinationUtils.getLatch(1, INIT_LATCH_ID);
-        LeaderElector leaderElector = coordinationUtils.getLeaderElector();
-        leaderElector.setLeaderElectorListener(() -> {
+        try {
+          Latch initLatch = coordinationUtils.getLatch(1, INIT_LATCH_ID);
+          LeaderElector leaderElector = coordinationUtils.getLeaderElector();
+          leaderElector.setLeaderElectorListener(() -> {
             getStreamManager().createStreams(intStreams);
             initLatch.countDown();
           });
-        leaderElector.tryBecomeLeader();
-        initLatch.await(LATCH_TIMEOUT_MINUTES, TimeUnit.MINUTES);
+          leaderElector.tryBecomeLeader();
+          initLatch.await(LATCH_TIMEOUT_MINUTES, TimeUnit.MINUTES);
+        } finally {
+          coordinationUtils.reset();
+        }
       } else {
         // each application process will try creating the streams, which
         // requires stream creation to be idempotent

From e5464499bbe4b01a5b0d758789974505802c1334 Mon Sep 17 00:00:00 2001
From: Bharath Kumarasubramanian <bkumaras@linkedin.com>
Date: Wed, 9 Aug 2017 16:25:41 -0700
Subject: [PATCH 3/7] Fix checkstyle issues and address PR comments

---
 .../org/apache/samza/runtime/LocalApplicationRunner.java  | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
index 20e26de66c..3fbd78faed 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
@@ -60,7 +60,7 @@ public class LocalApplicationRunner extends AbstractApplicationRunner {
   private static final Logger LOG = LoggerFactory.getLogger(LocalApplicationRunner.class);
   // Latch id that's used for awaiting the init of application before creating the StreamProcessors
   private static final String INIT_LATCH_ID = "init";
-  private static final String APPLICATION_RUNNER_ZK_PATH_SUFFIX = "/" + "ApplicationRunnerData";
+  private static final String APPLICATION_RUNNER_ZK_PATH_SUFFIX = "/ApplicationRunnerData";
   // Latch timeout is set to 10 min
   private static final long LATCH_TIMEOUT_MINUTES = 10;
 
@@ -236,9 +236,9 @@ public void waitForFinish() {
           Latch initLatch = coordinationUtils.getLatch(1, INIT_LATCH_ID);
           LeaderElector leaderElector = coordinationUtils.getLeaderElector();
           leaderElector.setLeaderElectorListener(() -> {
-            getStreamManager().createStreams(intStreams);
-            initLatch.countDown();
-          });
+              getStreamManager().createStreams(intStreams);
+              initLatch.countDown();
+            });
           leaderElector.tryBecomeLeader();
           initLatch.await(LATCH_TIMEOUT_MINUTES, TimeUnit.MINUTES);
         } finally {

From ab57e0e9c7d3bf5b6114314baed6139a91df88e1 Mon Sep 17 00:00:00 2001
From: Bharath Kumarasubramanian <bkumaras@linkedin.com>
Date: Thu, 10 Aug 2017 11:47:31 -0700
Subject: [PATCH 4/7] Leaving out the lifecycle problem and fixing only the
 path conflict

---
 .../samza/runtime/LocalApplicationRunner.java | 26 +++++++++----------
 1 file changed, 13 insertions(+), 13 deletions(-)

diff --git a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
index 3fbd78faed..621e99f38a 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
@@ -65,6 +65,7 @@ public class LocalApplicationRunner extends AbstractApplicationRunner {
   private static final long LATCH_TIMEOUT_MINUTES = 10;
 
   private final String uid;
+  private final CoordinationUtils coordinationUtils;
   private final Set<StreamProcessor> processors = ConcurrentHashMap.newKeySet();
   private final CountDownLatch shutdownLatch = new CountDownLatch(1);
   private final AtomicInteger numProcessorsToStart = new AtomicInteger();
@@ -123,6 +124,9 @@ private void shutdownAndNotify() {
         }
       }
 
+      if (coordinationUtils != null) {
+        coordinationUtils.reset();
+      }
       shutdownLatch.countDown();
     }
   }
@@ -130,6 +134,7 @@ private void shutdownAndNotify() {
   public LocalApplicationRunner(Config config) {
     super(config);
     uid = UUID.randomUUID().toString();
+    coordinationUtils = createCoordinationUtils();
   }
 
   @Override
@@ -230,20 +235,15 @@ public void waitForFinish() {
    */
   /* package private */ void createStreams(List<StreamSpec> intStreams) throws TimeoutException {
     if (!intStreams.isEmpty()) {
-      CoordinationUtils coordinationUtils = createCoordinationUtils();
       if (coordinationUtils != null) {
-        try {
-          Latch initLatch = coordinationUtils.getLatch(1, INIT_LATCH_ID);
-          LeaderElector leaderElector = coordinationUtils.getLeaderElector();
-          leaderElector.setLeaderElectorListener(() -> {
-              getStreamManager().createStreams(intStreams);
-              initLatch.countDown();
-            });
-          leaderElector.tryBecomeLeader();
-          initLatch.await(LATCH_TIMEOUT_MINUTES, TimeUnit.MINUTES);
-        } finally {
-          coordinationUtils.reset();
-        }
+        Latch initLatch = coordinationUtils.getLatch(1, INIT_LATCH_ID);
+        LeaderElector leaderElector = coordinationUtils.getLeaderElector();
+        leaderElector.setLeaderElectorListener(() -> {
+            getStreamManager().createStreams(intStreams);
+            initLatch.countDown();
+          });
+        leaderElector.tryBecomeLeader();
+        initLatch.await(LATCH_TIMEOUT_MINUTES, TimeUnit.MINUTES);
       } else {
         // each application process will try creating the streams, which
         // requires stream creation to be idempotent

From ec0b71810dfc5a25e3862e545a67c87b0a1ce4d0 Mon Sep 17 00:00:00 2001
From: Bharath Kumarasubramanian <bkumaras@linkedin.com>
Date: Fri, 11 Aug 2017 14:44:30 -0700
Subject: [PATCH 5/7] Fix the zookeeper idle connection problem & other bugs

---
 .../samza/runtime/LocalApplicationRunner.java | 81 +++++++++++++++----
 .../samza/zk/ZkJobCoordinatorFactory.java     |  3 +-
 .../org/apache/samza/zk/ZkProcessorLatch.java | 10 ++-
 3 files changed, 75 insertions(+), 19 deletions(-)

diff --git a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
index 621e99f38a..15812e1454 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
@@ -19,6 +19,7 @@
 
 package org.apache.samza.runtime;
 
+import com.google.common.base.Objects;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Set;
@@ -29,6 +30,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 import org.apache.samza.SamzaException;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.ApplicationConfig;
@@ -58,14 +60,12 @@
 public class LocalApplicationRunner extends AbstractApplicationRunner {
 
   private static final Logger LOG = LoggerFactory.getLogger(LocalApplicationRunner.class);
-  // Latch id that's used for awaiting the init of application before creating the StreamProcessors
-  private static final String INIT_LATCH_ID = "init";
   private static final String APPLICATION_RUNNER_ZK_PATH_SUFFIX = "/ApplicationRunnerData";
   // Latch timeout is set to 10 min
   private static final long LATCH_TIMEOUT_MINUTES = 10;
+  private static final long LEADER_ELECTION_WAIT_TIME_MS = 1000;
 
   private final String uid;
-  private final CoordinationUtils coordinationUtils;
   private final Set<StreamProcessor> processors = ConcurrentHashMap.newKeySet();
   private final CountDownLatch shutdownLatch = new CountDownLatch(1);
   private final AtomicInteger numProcessorsToStart = new AtomicInteger();
@@ -124,9 +124,6 @@ private void shutdownAndNotify() {
         }
       }
 
-      if (coordinationUtils != null) {
-        coordinationUtils.reset();
-      }
       shutdownLatch.countDown();
     }
   }
@@ -134,7 +131,6 @@ private void shutdownAndNotify() {
   public LocalApplicationRunner(Config config) {
     super(config);
     uid = UUID.randomUUID().toString();
-    coordinationUtils = createCoordinationUtils();
   }
 
   @Override
@@ -161,6 +157,7 @@ public void run(StreamApplication app) {
     try {
       // 1. initialize and plan
       ExecutionPlan plan = getExecutionPlan(app);
+
       writePlanJsonFile(plan.getPlanAsJson());
 
       // 2. create the necessary streams
@@ -231,19 +228,32 @@ public void waitForFinish() {
    * will create the streams. All the runner processes will wait on the latch that is released after the leader finishes
    * stream creation.
    * @param intStreams list of intermediate {@link StreamSpec}s
-   * @throws Exception exception for latch timeout
+   * @throws TimeoutException exception for latch timeout
    */
   /* package private */ void createStreams(List<StreamSpec> intStreams) throws TimeoutException {
     if (!intStreams.isEmpty()) {
+      // Move the scope of coordination utils within stream creation to address long idle connection problem.
+      // Refer SAMZA-1385 for more details
+      CoordinationUtils coordinationUtils = createCoordinationUtils();
       if (coordinationUtils != null) {
-        Latch initLatch = coordinationUtils.getLatch(1, INIT_LATCH_ID);
-        LeaderElector leaderElector = coordinationUtils.getLeaderElector();
-        leaderElector.setLeaderElectorListener(() -> {
-            getStreamManager().createStreams(intStreams);
-            initLatch.countDown();
-          });
-        leaderElector.tryBecomeLeader();
-        initLatch.await(LATCH_TIMEOUT_MINUTES, TimeUnit.MINUTES);
+        // generate a unique reproducible id for an application lifecycle
+        String latchId = generateLatchId(intStreams);
+        Latch initLatch = coordinationUtils.getLatch(1, latchId);
+
+        try {
+          // check if the processor needs to go through leader election and stream creation
+          if (shouldContestInElectionForStreamCreation(initLatch)) {
+            LeaderElector leaderElector = coordinationUtils.getLeaderElector();
+            leaderElector.setLeaderElectorListener(() -> {
+              getStreamManager().createStreams(intStreams);
+              initLatch.countDown();
+            });
+            leaderElector.tryBecomeLeader();
+            initLatch.await(LATCH_TIMEOUT_MINUTES, TimeUnit.MINUTES);
+          }
+        } finally {
+          coordinationUtils.reset();
+        }
       } else {
         // each application process will try creating the streams, which
         // requires stream creation to be idempotent
@@ -275,4 +285,43 @@ StreamProcessor createStreamProcessor(
           taskFactory.getClass().getCanonicalName()));
     }
   }
+
+  /**
+   * Generates a unique latch id that is consistent across processors within the same application lifecycle.
+   * Each {@code StreamSpec} has an id that is unique and is used for hashcode computation.
+   *
+   * @param intStreams list of {@link StreamSpec}s
+   * @return latch id
+   */
+  private String generateLatchId(List<StreamSpec> intStreams) {
+    return String.valueOf(Objects.hashCode(intStreams.stream()
+        .map(StreamSpec::getId)
+        .collect(Collectors.toList())));
+  }
+
+  /**
+   * By closing the zk connection, we clean up the ephemeral node used for leader election. It results in a leader election
+   * whenever a new process joins and also creates intermediate streams.
+   *    1. It is unnecessary to re-conduct leader election for stream creation in the same application lifecycle
+   *    2. Underlying systems may not support check for stream existence prior to creation which could have potential problems.
+   * As in interim solution, we reuse the same latch as a marker to determine if create streams phase is done for the
+   * application lifecycle using {@link Latch#await(long, TimeUnit)}
+   *
+   * @param streamCreationLatch latch used for stream creation
+   * @return true if processor needs to be part of election
+   *         false otherwise
+   */
+  private boolean shouldContestInElectionForStreamCreation(Latch streamCreationLatch) {
+    boolean eligibleForElection = true;
+
+    try {
+      streamCreationLatch.await(LEADER_ELECTION_WAIT_TIME_MS, TimeUnit.MILLISECONDS);
+      // case we didn't time out suggesting that latch already exists
+      eligibleForElection = false;
+    } catch(TimeoutException e) {
+      LOG.info("Timed out waiting for the latch! Going to enter leader election section to create streams");
+    }
+
+    return eligibleForElection;
+  }
 }
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
index 85e3b4ab55..7c12005308 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
@@ -33,6 +33,7 @@
 public class ZkJobCoordinatorFactory implements JobCoordinatorFactory {
 
   private static final Logger LOG = LoggerFactory.getLogger(ZkJobCoordinatorFactory.class);
+  private static final String JOB_COORDINATOR_ZK_PATH_SUFFIX = "/JobCoordinatorData";
 
   /**
    * Method to instantiate an implementation of JobCoordinator
@@ -50,7 +51,7 @@ public JobCoordinator getJobCoordinator(Config config) {
 
   private ZkUtils getZkUtils(Config config, MetricsRegistry metricsRegistry) {
     ZkConfig zkConfig = new ZkConfig(config);
-    ZkKeyBuilder keyBuilder = new ZkKeyBuilder(new ApplicationConfig(config).getGlobalAppId());
+    ZkKeyBuilder keyBuilder = new ZkKeyBuilder(new ApplicationConfig(config).getGlobalAppId() + JOB_COORDINATOR_ZK_PATH_SUFFIX);
     ZkClient zkClient = ZkCoordinationServiceFactory.createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
     return new ZkUtils(keyBuilder, zkClient, zkConfig.getZkConnectionTimeoutMs(), metricsRegistry);
   }
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java b/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java
index ddc897665a..83fe38963e 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java
@@ -20,6 +20,8 @@
 
 import java.util.concurrent.TimeUnit;
 
+import java.util.concurrent.TimeoutException;
+import org.I0Itec.zkclient.exception.ZkInterruptedException;
 import org.apache.samza.coordinator.Latch;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -53,8 +55,12 @@ public ZkProcessorLatch(int size, String latchId, String participantId, ZkUtils
   }
 
   @Override
-  public void await(long timeout, TimeUnit tu) {
-    zkUtils.getZkClient().waitUntilExists(targetPath, TimeUnit.MILLISECONDS, timeout);
+  public void await(long timeout, TimeUnit tu) throws TimeoutException {
+    boolean targetPathExists = zkUtils.getZkClient().waitUntilExists(targetPath, TimeUnit.MILLISECONDS, timeout);
+
+    if (!targetPathExists) {
+      throw new TimeoutException("Timed out waiting for the targetPath");
+    }
   }
 
   @Override

From 4a4bfc784076c071c9b4c12266966d0bd65f03f5 Mon Sep 17 00:00:00 2001
From: Bharath Kumarasubramanian <bkumaras@linkedin.com>
Date: Fri, 11 Aug 2017 15:16:15 -0700
Subject: [PATCH 6/7] Fix docs and remove unused import

---
 .../org/apache/samza/runtime/LocalApplicationRunner.java     | 5 +++--
 .../src/main/java/org/apache/samza/zk/ZkProcessorLatch.java  | 1 -
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
index 15812e1454..b27e496de7 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
@@ -300,8 +300,9 @@ private String generateLatchId(List<StreamSpec> intStreams) {
   }
 
   /**
-   * By closing the zk connection, we clean up the ephemeral node used for leader election. It results in a leader election
-   * whenever a new process joins and also creates intermediate streams.
+   * In order to fix SAMZA-1385, we are limiting the scope of coordination util within stream creation phase and destroying
+   * the coordination util right after. By closing the zk connection, we clean up the ephemeral node used for leader election.
+   * It results in a leader election whenever a new process joins and also creates intermediate streams.
    *    1. It is unnecessary to re-conduct leader election for stream creation in the same application lifecycle
    *    2. Underlying systems may not support check for stream existence prior to creation which could have potential problems.
    * As in interim solution, we reuse the same latch as a marker to determine if create streams phase is done for the
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java b/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java
index 83fe38963e..f8960cb5bb 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java
@@ -21,7 +21,6 @@
 import java.util.concurrent.TimeUnit;
 
 import java.util.concurrent.TimeoutException;
-import org.I0Itec.zkclient.exception.ZkInterruptedException;
 import org.apache.samza.coordinator.Latch;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

From be60eefae7b1f055a40a1d84225c2e142f539b69 Mon Sep 17 00:00:00 2001
From: Bharath Kumarasubramanian <bkumaras@linkedin.com>
Date: Mon, 14 Aug 2017 16:40:39 -0700
Subject: [PATCH 7/7] Address Navina's comments on multi stage job and use plan
 hashcode instead of custom generated hash

---
 .../samza/runtime/LocalApplicationRunner.java | 33 ++------
 .../samza/zk/ZkJobCoordinatorFactory.java     | 21 ++++-
 .../org/apache/samza/zk/ZkProcessorLatch.java |  2 +-
 .../runtime/TestLocalApplicationRunner.java   | 82 ++++++++++++++-----
 4 files changed, 90 insertions(+), 48 deletions(-)

diff --git a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
index fb95b39f3d..588e657f0d 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
@@ -19,7 +19,6 @@
 
 package org.apache.samza.runtime;
 
-import com.google.common.base.Objects;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Set;
@@ -30,7 +29,6 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Collectors;
 import org.apache.samza.SamzaException;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.ApplicationConfig;
@@ -158,10 +156,13 @@ public void run(StreamApplication app) {
       // 1. initialize and plan
       ExecutionPlan plan = getExecutionPlan(app);
 
-      writePlanJsonFile(plan.getPlanAsJson());
+      String executionPlanJson = plan.getPlanAsJson();
+      writePlanJsonFile(executionPlanJson);
 
       // 2. create the necessary streams
-      createStreams(plan.getIntermediateStreams());
+      // TODO: System generated intermediate streams should have robust naming scheme. Refer JIRA-1391
+      String planId = String.valueOf(executionPlanJson.hashCode());
+      createStreams(planId, plan.getIntermediateStreams());
 
       // 3. create the StreamProcessors
       if (plan.getJobConfigs().isEmpty()) {
@@ -227,18 +228,17 @@ public void waitForFinish() {
    * If {@link CoordinationUtils} is provided, this function will first invoke leader election, and the leader
    * will create the streams. All the runner processes will wait on the latch that is released after the leader finishes
    * stream creation.
+   * @param planId a unique identifier representing the plan used for coordination purpose
    * @param intStreams list of intermediate {@link StreamSpec}s
    * @throws TimeoutException exception for latch timeout
    */
-  /* package private */ void createStreams(List<StreamSpec> intStreams) throws TimeoutException {
+  /* package private */ void createStreams(String planId, List<StreamSpec> intStreams) throws TimeoutException {
     if (!intStreams.isEmpty()) {
       // Move the scope of coordination utils within stream creation to address long idle connection problem.
       // Refer SAMZA-1385 for more details
       CoordinationUtils coordinationUtils = createCoordinationUtils();
       if (coordinationUtils != null) {
-        // generate a unique reproducible id for an application lifecycle
-        String latchId = generateLatchId(intStreams);
-        Latch initLatch = coordinationUtils.getLatch(1, latchId);
+        Latch initLatch = coordinationUtils.getLatch(1, planId);
 
         try {
           // check if the processor needs to go through leader election and stream creation
@@ -286,23 +286,6 @@ StreamProcessor createStreamProcessor(
     }
   }
 
-  /**
-   * Generates a unique latch id that is consistent across processors within the same application lifecycle.
-   * Each {@code StreamSpec} has an id that is unique and is used for hashcode computation.
-   *
-   * @param intStreams list of {@link StreamSpec}s
-   * @return latch id
-   */
-  /* package private */
-  String generateLatchId(List<StreamSpec> intStreams) {
-    return String.valueOf(
-        Objects.hashCode(
-            intStreams.stream()
-                .map(StreamSpec::getId)
-                .sorted()
-                .collect(Collectors.toList())));
-  }
-
   /**
    * In order to fix SAMZA-1385, we are limiting the scope of coordination util within stream creation phase and destroying
    * the coordination util right after. By closing the zk connection, we clean up the ephemeral node used for leader election.
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
index 7c12005308..08d826ec60 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
@@ -22,6 +22,7 @@
 import org.I0Itec.zkclient.ZkClient;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
+import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.ZkConfig;
 import org.apache.samza.coordinator.JobCoordinator;
 import org.apache.samza.coordinator.JobCoordinatorFactory;
@@ -30,10 +31,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
 public class ZkJobCoordinatorFactory implements JobCoordinatorFactory {
 
   private static final Logger LOG = LoggerFactory.getLogger(ZkJobCoordinatorFactory.class);
-  private static final String JOB_COORDINATOR_ZK_PATH_SUFFIX = "/JobCoordinatorData";
+  private static final String JOB_COORDINATOR_ZK_PATH_FORMAT = "%s/%s-%s-coordinationData";
+  private static final String DEFAULT_JOB_ID = "1";
+  private static final String DEFAULT_JOB_NAME = "defaultJob";
 
   /**
    * Method to instantiate an implementation of JobCoordinator
@@ -51,8 +55,21 @@ public JobCoordinator getJobCoordinator(Config config) {
 
   private ZkUtils getZkUtils(Config config, MetricsRegistry metricsRegistry) {
     ZkConfig zkConfig = new ZkConfig(config);
-    ZkKeyBuilder keyBuilder = new ZkKeyBuilder(new ApplicationConfig(config).getGlobalAppId() + JOB_COORDINATOR_ZK_PATH_SUFFIX);
+    ZkKeyBuilder keyBuilder = new ZkKeyBuilder(getJobCoordinationZkPath(config));
     ZkClient zkClient = ZkCoordinationServiceFactory.createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
     return new ZkUtils(keyBuilder, zkClient, zkConfig.getZkConnectionTimeoutMs(), metricsRegistry);
   }
+
+  private String getJobCoordinationZkPath(Config config) {
+    JobConfig jobConfig = new JobConfig(config);
+    String appId = new ApplicationConfig(config).getGlobalAppId();
+    String jobName = jobConfig.getName().isDefined()
+        ? jobConfig.getName().get()
+        : DEFAULT_JOB_NAME;
+    String jobId = jobConfig.getJobId().isDefined()
+        ? jobConfig.getJobId().get()
+        : DEFAULT_JOB_ID;
+
+    return String.format(JOB_COORDINATOR_ZK_PATH_FORMAT, appId, jobName, jobId);
+  }
 }
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java b/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java
index 597297a780..166c6276c2 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkProcessorLatch.java
@@ -55,7 +55,7 @@ public ZkProcessorLatch(int size, String latchId, String participantId, ZkUtils
 
   @Override
   public void await(long timeout, TimeUnit timeUnit) throws TimeoutException {
-    // waitUntilExists signals timeout is by returning false as opposed to throwing exception. We internally need to map
+    // waitUntilExists signals timeout by returning false as opposed to throwing exception. We internally need to map
     // the non-existence to a TimeoutException in order to respect the contract defined in Latch interface
     boolean targetPathExists = zkUtils.getZkClient().waitUntilExists(targetPath, timeUnit, timeout);
 
diff --git a/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java b/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java
index 9188de95ea..6c7827e9ac 100644
--- a/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java
+++ b/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java
@@ -27,9 +27,9 @@
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.ApplicationConfig;
-import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.TaskConfig;
@@ -56,6 +56,22 @@
 
 public class TestLocalApplicationRunner {
 
+  private static final String PLAN_JSON = "{"
+      + "\"jobs\":[{"
+      + "\"jobName\":\"test-application\","
+      + "\"jobId\":\"1\","
+      + "\"operatorGraph\":{"
+      + "\"intermediateStreams\":{%s},"
+      + "\"applicationName\":\"test-application\",\"applicationId\":\"1\"}";
+  private static final String STREAM_SPEC_JSON_FORMAT = "\"%s\":{"
+      + "\"streamSpec\":{"
+      + "\"id\":\"%s\","
+      + "\"systemName\":\"%s\","
+      + "\"physicalName\":\"%s\","
+      + "\"partitionCount\":2},"
+      + "\"sourceJobs\":[\"test-app\"],"
+      + "\"targetJobs\":[\"test-target-app\"]},";
+
   @Test
   public void testStreamCreation() throws Exception {
     Map<String, String> config = new HashMap<>();
@@ -173,8 +189,10 @@ public boolean amILeader() {
       @Override
       public void await(long timeout, TimeUnit tu)
           throws TimeoutException {
-        // in this test, latch is released before wait
-        assertTrue(done);
+        // in this test, latch is released after countDown is invoked
+        if (!done) {
+          throw new TimeoutException("timed out waiting for the target path");
+        }
       }
 
       @Override
@@ -336,56 +354,80 @@ public String getPlanAsJson()
     assertEquals(spy.status(app), ApplicationStatus.UnsuccessfulFinish);
   }
 
+  /**
+   * A test case to verify if the plan results in different hash if there is change in topological sort order.
+   * Note: the overall JOB PLAN remains the same outside the scope of intermediate streams the sake of these test cases.
+   */
   @Test
-  public void testGenerateLatchIdWithShuffledStreamSpecs() {
+  public void testPlanIdWithShuffledStreamSpecs() {
     List<StreamSpec> streamSpecs = ImmutableList.of(
       new StreamSpec("test-stream-1", "stream-1", "testStream"),
         new StreamSpec("test-stream-2", "stream-2", "testStream"),
         new StreamSpec("test-stream-3", "stream-3", "testStream"));
-    String latchIdBeforeShuffle = generateLatchId(streamSpecs);
+    String planIdBeforeShuffle = getExecutionPlanId(streamSpecs);
 
     List<StreamSpec> shuffledStreamSpecs = ImmutableList.of(
         new StreamSpec("test-stream-2", "stream-2", "testStream"),
         new StreamSpec("test-stream-1", "stream-1", "testStream"),
         new StreamSpec("test-stream-3", "stream-3", "testStream"));
 
-    assertEquals("Expected both of the latch ids to be same", latchIdBeforeShuffle,
-        generateLatchId(shuffledStreamSpecs));
+    assertNotEquals("Expected both of the latch ids to be different", planIdBeforeShuffle,
+        getExecutionPlanId(shuffledStreamSpecs));
   }
 
+  /**
+   * A test case to verify if the plan results in same hash in case of same plan.
+   * Note: the overall JOB PLAN remains the same outside the scope of intermediate streams the sake of these test cases.
+   */
   @Test
-  public void testGenerateLatchIdWithSameStreamSpecs() {
+  public void testGeneratePlanIdWithSameStreamSpecs() {
     List<StreamSpec> streamSpecs = ImmutableList.of(
         new StreamSpec("test-stream-1", "stream-1", "testStream"),
         new StreamSpec("test-stream-2", "stream-2", "testStream"),
         new StreamSpec("test-stream-3", "stream-3", "testStream"));
-    String latchIdForFirstAttempt = generateLatchId(streamSpecs);
-    String latchIdForSecondAttempt = generateLatchId(streamSpecs);
+    String planIdForFirstAttempt = getExecutionPlanId(streamSpecs);
+    String planIdForSecondAttempt = getExecutionPlanId(streamSpecs);
 
-    assertEquals("Expected latch ids to match!", "-1037489986", latchIdForFirstAttempt);
-    assertEquals("Expected latch ids to match for the second attempt!", latchIdForFirstAttempt, latchIdForSecondAttempt);
+    assertEquals("Expected latch ids to match!", "1447946713", planIdForFirstAttempt);
+    assertEquals("Expected latch ids to match for the second attempt!", planIdForFirstAttempt, planIdForSecondAttempt);
   }
 
+  /**
+   * A test case to verify plan results in different hash in case of different intermediate stream.
+   * Note: the overall JOB PLAN remains the same outside the scope of intermediate streams the sake of these test cases.
+   */
   @Test
-  public void testGenerateLatchIdWithDifferentStreamSpecs() {
+  public void testGeneratePlanIdWithDifferentStreamSpecs() {
     List<StreamSpec> streamSpecs = ImmutableList.of(
         new StreamSpec("test-stream-1", "stream-1", "testStream"),
         new StreamSpec("test-stream-2", "stream-2", "testStream"),
         new StreamSpec("test-stream-3", "stream-3", "testStream"));
-    String latchIdBeforeShuffle = generateLatchId(streamSpecs);
+    String planIdBeforeShuffle = getExecutionPlanId(streamSpecs);
 
-    List<StreamSpec> shuffledStreamSpecs = ImmutableList.of(
+    List<StreamSpec> updatedStreamSpecs = ImmutableList.of(
         new StreamSpec("test-stream-1", "stream-1", "testStream"),
         new StreamSpec("test-stream-4", "stream-4", "testStream"),
         new StreamSpec("test-stream-3", "stream-3", "testStream"));
 
-    assertNotEquals("Expected both of the latch ids to be different", latchIdBeforeShuffle,
-        generateLatchId(shuffledStreamSpecs));
+    assertNotEquals("Expected both of the latch ids to be different", planIdBeforeShuffle,
+        getExecutionPlanId(updatedStreamSpecs));
   }
 
-  private String generateLatchId(List<StreamSpec> streamSpecs) {
-    LocalApplicationRunner runner = new LocalApplicationRunner(mock(Config.class));
-    return runner.generateLatchId(streamSpecs);
+  private String getExecutionPlanId(List<StreamSpec> updatedStreamSpecs) {
+    String intermediateStreamJson = updatedStreamSpecs.stream()
+        .map(this::streamSpecToJson)
+        .collect(Collectors.joining(","));
+
+    int planId = String.format(PLAN_JSON, intermediateStreamJson).hashCode();
+
+    return String.valueOf(planId);
   }
 
+  private String streamSpecToJson(StreamSpec streamSpec) {
+    return String.format(STREAM_SPEC_JSON_FORMAT,
+        streamSpec.getId(),
+        streamSpec.getId(),
+        streamSpec.getSystemName(),
+        streamSpec.getPhysicalName());
+  }
 }