YARN-11656 RMStateStore event queue blocked

- fix style

Author: K0K0V0K

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/event/multidispatcher/MultiDispatcher.java

@@ -43,6 +43,8 @@
  * Dispatches {@link Event}s in a parallel thread.
  * The {@link Dispatcher#getEventHandler()} method can be used to post an event to the dispatcher.
  * The posted event will be separated based on the hashcode of the {@link Event#getLockKey()}.
+ * If the getLockKey() method returns with null,
+ * then the first executor thread will be used as default worker
  * The {@link MultiDispatcherConfig} contains the information,
  * how many thread will be used, for parallel execution.
  * The {@link MultiDispatcherExecutor} contains the worker threads, which handle the events.
@@ -65,7 +67,6 @@ public MultiDispatcher(String dispatcherName) {
     super("Dispatcher");
     this.dispatcherName = dispatcherName.replaceAll(" ", "-").toLowerCase();
     this.log = LoggerFactory.getLogger(MultiDispatcher.class.getCanonicalName() + "." + this.dispatcherName);
-    this.metrics = new DispatcherEventMetricsImpl(this.dispatcherName);
     this.library = new MultiDispatcherLibrary();
   }
 
@@ -74,6 +75,7 @@ protected void serviceInit(Configuration conf) throws Exception{
     super.serviceInit(conf);
     MultiDispatcherConfig config = new MultiDispatcherConfig(getConfig(), dispatcherName);
     workerExecutor = new MultiDispatcherExecutor(log, config, dispatcherName);
+    workerExecutor.start();
     createMonitorThread(config);
     if (config.getMetricsEnabled()) {
       metrics = new DispatcherEventMetricsImpl(dispatcherName);
@@ -89,6 +91,9 @@ protected void serviceInit(Configuration conf) throws Exception{
 
   @Override
   protected void serviceStop() throws Exception {
+    if (monitorExecutor != null) {
+      monitorExecutor.shutdown();
+    }
     workerExecutor.stop();
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/event/multidispatcher/MultiDispatcherConfig.java

@@ -32,22 +32,45 @@ public MultiDispatcherConfig(Configuration configuration, String dispatcherName)
     this.prefix = String.format("yarn.dispatcher.multi-thread.%s.", dispatcherName);
   }
 
+  /**
+   * How many executor thread should be created to handle the incoming events
+   * @return configured value, or default 4
+   */
   public int getDefaultPoolSize() {
     return super.getInt(prefix + "default-pool-size", 4);
   }
 
+  /**
+   * Maximus size of the event queue of the executor threads.
+   * If limit is reached then the queue#add method will block.
+   * @return configured value, or default 1_000_000
+   */
   public int getQueueSize() {
-    return super.getInt(prefix + "queue-size", 10_000_000);
+    return super.getInt(prefix + "queue-size", 1_000_000);
   }
 
+  /**
+   * How frequently the monitor thread should write the state of the dispatcher to the LOG.
+   * If less than 1 this monitoring will be turned off.
+   * @return configured value, or default 0
+   */
   public int getMonitorSeconds() {
-    return super.getInt(prefix + "monitor-seconds", 30);
+    return super.getInt(prefix + "monitor-seconds", 0);
   }
 
+  /**
+   * How long should the dispatcher wait to drain all event queue of workers,
+   * after stop signal is received.
+   * @return configured value, or default 60
+   */
   public int getGracefulStopSeconds() {
     return super.getInt(prefix + "graceful-stop-seconds", 60);
   }
 
+  /**
+   * Dispatcher metrics should be published to the metric system.
+   * @return configured value, or default false
+   */
   public boolean getMetricsEnabled() {
     return super.getBoolean(prefix + "metrics-enabled", false);
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/event/multidispatcher/MultiDispatcherExecutor.java

@@ -35,32 +35,39 @@
  */
 public class MultiDispatcherExecutor {
 
-  private final Logger LOG;
+  private final Logger log;
   private final MultiDispatcherConfig config;
   private final MultiDispatcherExecutorThread[] threads;
   private final Clock clock = new MonotonicClock();
 
   public MultiDispatcherExecutor(
-      Logger LOG,
+      Logger log,
       MultiDispatcherConfig config,
       String dispatcherName
   ) {
-    this.LOG = LOG;
+    this.log = log;
     this.config = config;
     this.threads = new MultiDispatcherExecutorThread[config.getDefaultPoolSize()];
     ThreadGroup group = new ThreadGroup(dispatcherName);
-    for (int i=0; i < threads.length; ++i) {
+    for (int i = 0; i < threads.length; ++i) {
       threads[i] = new MultiDispatcherExecutorThread(group, i, config.getQueueSize());
-      threads[i].start();
+    }
+  }
+
+  public void start() {
+    for(Thread t : threads) {
+      t.start();
     }
   }
 
   public void execute(Event event, Runnable runnable) {
     String lockKey = event.getLockKey();
-    int threadIndex = lockKey == null ? 0 : Math.abs(lockKey.hashCode()) % threads.length;
+    // abs of Integer.MIN_VALUE is Integer.MIN_VALUE
+    int threadIndex = lockKey == null  || lockKey.hashCode() == Integer.MIN_VALUE ?
+        0 : Math.abs(lockKey.hashCode() % threads.length);
     MultiDispatcherExecutorThread thread = threads[threadIndex];
     thread.add(runnable);
-    LOG.debug("The {} with lock key {} will be handled by {}",
+    log.trace("The {} with lock key {} will be handled by {}",
         event.getType(), lockKey, thread.getName());
   }
 
@@ -70,7 +77,7 @@ public void stop() throws InterruptedException {
     if (Arrays.stream(threads).anyMatch(t -> 0 < t.queueSize())
         // and not timeout yet
       && clock.getTime() < timeOut) {
-      LOG.debug("Not all event queue is empty, waiting to drain ...");
+      log.debug("Not all event queue is empty, waiting to drain ...");
       Thread.sleep(1_000);
     }
     for (MultiDispatcherExecutorThread thread : threads) {
@@ -108,7 +115,7 @@ public void run() {
           queue.take().run();
         }
       } catch (InterruptedException e) {
-        LOG.warn("{} get interrupted", getName());
+        log.warn("{} get interrupted", getName());
       }
     }
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/metrics/DispatcherEventMetrics.java

@@ -19,13 +19,34 @@
 
 import org.apache.hadoop.metrics2.MetricsSource;
 
+/**
+ * Interface for {@link org.apache.hadoop.yarn.event.Dispatcher}
+ * can be used to publish {@link org.apache.hadoop.yarn.event.Event} related metrics
+ */
 public interface DispatcherEventMetrics extends MetricsSource {
 
+  /**
+   * Class of the event type what can be handled by the DispatcherEventMetrics
+   * @param typeClass the event type
+   */
   void init(Class<? extends Enum> typeClass);
 
+  /**
+   * Call if Event added for dispatching
+   * @param type type of the event
+   */
   void addEvent(Object type);
 
+  /**
+   * Call if Event handled
+   * @param type type of the event
+   */
   void removeEvent(Object type);
 
+  /**
+   * Call with how much time was required to handle the event
+   * @param type type of the event
+   * @param millisecond time interval
+   */
   void updateRate(Object type, long millisecond);
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/metrics/DispatcherEventMetricsImpl.java

@@ -29,6 +29,9 @@
 import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
 import org.apache.hadoop.metrics2.lib.MutableRate;
 
+/**
+ * Metric object for {@link org.apache.hadoop.yarn.event.multidispatcher.MultiDispatcher}
+ */
 @InterfaceAudience.Private
 @Metrics(context="yarn")
 public class DispatcherEventMetricsImpl implements DispatcherEventMetrics {
@@ -53,11 +56,12 @@ public void getMetrics(MetricsCollector collector, boolean all) {
 
   @Override
   public void init(Class<? extends Enum> typeClass) {
-    for(Object c : typeClass.getEnumConstants()) {
-      String key = createKey(c);
-      currentEventCountMetrics.put(key, this.registry.newGauge(
-          key + "_Current", key + "_Current", 0L));
-      processingTimeMetrics.put(key, this.registry.newRate(key + "_", key+ "_"));
+    for(Object constant : typeClass.getEnumConstants()) {
+      String key = createKey(constant);
+      currentEventCountMetrics.put(key,
+          registry.newGauge(key + "_Current", key + "_Current", 0L));
+      processingTimeMetrics.put(key,
+          registry.newRate(key + "_", key + "_"));
     }
   }
 
@@ -87,5 +91,4 @@ public String toString() {
         .add("processingTimeMetrics=" + processingTimeMetrics)
         .toString();
   }
-
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/metrics/DispatcherEventMetricsNoOps.java

@@ -21,6 +21,9 @@
 
 import org.apache.hadoop.metrics2.MetricsCollector;
 
+/**
+ * Used if metric publication should be disabled
+ */
 public class DispatcherEventMetricsNoOps implements DispatcherEventMetrics {
 
   private final Logger log;
@@ -31,26 +34,26 @@ public DispatcherEventMetricsNoOps(Logger log) {
 
   @Override
   public void getMetrics(MetricsCollector collector, boolean all) {
-    log.debug("called getMetrics");
+    log.trace("called getMetrics");
   }
 
   @Override
   public void init(Class<? extends Enum> typeClass) {
-    log.debug("called init");
+    log.trace("called init");
   }
 
   @Override
   public void addEvent(Object type) {
-    log.debug("called addEvent");
+    log.trace("called addEvent");
   }
 
   @Override
   public void removeEvent(Object type) {
-    log.debug("called removeEvent");
+    log.trace("called removeEvent");
   }
 
   @Override
   public void updateRate(Object type, long millisecond) {
-    log.debug("called updateRate");
+    log.trace("called updateRate");
   }
 }