SAMZA-1309; Debounce time config

Author: Boris Shkolnik <[email protected]>

Reviewers: Navina Ramesh <[email protected]>

Closes apache#203 from sborya/DebounceConfig

Author: sborya

docs/learn/documentation/versioned/jobs/configuration-table.html

@@ -461,10 +461,18 @@ <h1>Samza Configuration Reference</h1>
 			How long each processor will wait for all the processors to report acceptance of the new job model before rolling back.
                     </td>
                 </tr>
+                <tr>
+                    <td class="property" id="job.debounce.time.ms">job.debounce.time.ms</td>
+                    <td class="default"> 2000 </td>
+                    <td class="description">
+                        How long the Leader processor will wait before recalculating the JobModel on change of registered processors.
+                    </td>
+                </tr>
                 <tr>
                     <th colspan="3" class="section" id="task"><a href="../api/overview.html">Task configuration</a></th>
                 </tr>
 
+
                 <tr>
                     <td class="property" id="task-class">task.class</td>
                     <td class="default"></td>

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

@@ -48,8 +48,6 @@ public class ScheduleAfterDebounceTime {
   // Action name when the Processor membership changes
   public static final String ON_PROCESSOR_CHANGE = "OnProcessorChange";
 
-  public static final int DEBOUNCE_TIME_MS = 2000;
-
   private final ScheduledTaskFailureCallback scheduledTaskFailureCallback;
 
   private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(

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

@@ -18,9 +18,13 @@
  */
 package org.apache.samza.zk;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.ZkConfig;
 import org.apache.samza.coordinator.JobCoordinator;
 import org.apache.samza.coordinator.JobCoordinatorListener;
@@ -34,10 +38,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
 /**
  * JobCoordinator for stand alone processor managed via Zookeeper.
  */
@@ -47,6 +47,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
   // with locality. Since host-affinity is not yet implemented, this can be fixed as part of SAMZA-1197
   private static final int METADATA_CACHE_TTL_MS = 5000;
 
+
   private final ZkUtils zkUtils;
   private final String processorId;
   private final ZkController zkController;
@@ -59,6 +60,8 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
   private JobCoordinatorListener coordinatorListener = null;
   private JobModel newJobModel;
 
+  private int debounceTimeMs;
+
   public ZkJobCoordinator(Config config) {
     this.config = config;
     ZkConfig zkConfig = new ZkConfig(config);
@@ -79,11 +82,14 @@ public ZkJobCoordinator(Config config) {
         keyBuilder.getJobModelVersionBarrierPrefix(),
         zkUtils,
         new ZkBarrierListenerImpl());
+    this.debounceTimeMs = new JobConfig(config).getDebounceTimeMs();
+
   }
 
   @Override
   public void start() {
     streamMetadataCache = StreamMetadataCache.apply(METADATA_CACHE_TTL_MS, config);
+
     debounceTimer = new ScheduleAfterDebounceTime(throwable -> {
         LOG.error("Received exception from in JobCoordinator Processing!", throwable);
         stop();
@@ -126,7 +132,7 @@ public String getProcessorId() {
   public void onProcessorChange(List<String> processors) {
     LOG.info("ZkJobCoordinator::onProcessorChange - list of processors changed! List size=" + processors.size());
     debounceTimer.scheduleAfterDebounceTime(ScheduleAfterDebounceTime.ON_PROCESSOR_CHANGE,
-        ScheduleAfterDebounceTime.DEBOUNCE_TIME_MS, () -> doOnProcessorChange(processors));
+        debounceTimeMs, () -> doOnProcessorChange(processors));
   }
 
   void doOnProcessorChange(List<String> processors) {
@@ -232,8 +238,8 @@ public void onBecomingLeader() {
       zkController.subscribeToProcessorChange();
       debounceTimer.scheduleAfterDebounceTime(
         ScheduleAfterDebounceTime.ON_PROCESSOR_CHANGE,
-        ScheduleAfterDebounceTime.DEBOUNCE_TIME_MS, () -> {
-          // actual actions to do are the same as onProcessorChange()
+        debounceTimeMs, () -> {
+          // actual actions to do are the same as onProcessorChange
           doOnProcessorChange(new ArrayList<>());
         });
     }

samza-core/src/main/scala/org/apache/samza/config/JobConfig.scala

@@ -47,6 +47,8 @@ object JobConfig {
   val JOB_CONTAINER_THREAD_POOL_SIZE = "job.container.thread.pool.size"
   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 SSP_GROUPER_FACTORY = "job.systemstreampartition.grouper.factory"
 
@@ -172,4 +174,6 @@ class JobConfig(config: Config) extends ScalaMapConfig(config) with Logging {
     case Some(mode) => mode.toBoolean
     case _ => false
   }
+
+  def getDebounceTimeMs = getInt(JobConfig.JOB_DEBOUNCE_TIME_MS, JobConfig.DEFAULT_DEBOUNCE_TIME_MS)
 }