YARN-6342. Make TimelineV2Client's drain timeout after stop configurable (Haibo Chen via Varun Saxena)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java

@@ -2095,6 +2095,16 @@ public class YarnConfiguration extends Configuration {
 
   public static final int DEFAULT_NUMBER_OF_ASYNC_ENTITIES_TO_MERGE = 10;
 
+
+  /**
+   * The time period for which timeline v2 client will wait for draining
+   * leftover entities after stop.
+   */
+  public static final String TIMELINE_V2_CLIENT_DRAIN_TIME_MILLIS =
+      TIMELINE_SERVICE_CLIENT_PREFIX + "drain-entities.timeout.ms";
+  public static final long DEFAULT_TIMELINE_V2_CLIENT_DRAIN_TIME_MILLIS
+      = 2000L;
+
   // mark app-history related configs @Private as application history is going
   // to be integrated into the timeline service
   @Private

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineV2ClientImpl.java

@@ -289,7 +289,7 @@ public class TimelineV2ClientImpl extends TimelineV2Client {
      * Time period for which the timelineclient will wait for draining after
      * stop.
      */
-    private static final long DRAIN_TIME_PERIOD = 2000L;
+    private final long drainTimeoutPeriod;
 
     private int numberOfAsyncsToMerge;
     private final BlockingQueue<EntitiesHolder> timelineEntityQueue;
@@ -300,6 +300,9 @@ public class TimelineV2ClientImpl extends TimelineV2Client {
       numberOfAsyncsToMerge =
           conf.getInt(YarnConfiguration.NUMBER_OF_ASYNC_ENTITIES_TO_MERGE,
               YarnConfiguration.DEFAULT_NUMBER_OF_ASYNC_ENTITIES_TO_MERGE);
+      drainTimeoutPeriod = conf.getLong(
+          YarnConfiguration.TIMELINE_V2_CLIENT_DRAIN_TIME_MILLIS,
+          YarnConfiguration.DEFAULT_TIMELINE_V2_CLIENT_DRAIN_TIME_MILLIS);
     }
 
     Runnable createRunnable() {
@@ -330,7 +333,7 @@ public class TimelineV2ClientImpl extends TimelineV2Client {
             // Try to drain the remaining entities to be published @ the max for
             // 2 seconds
             long timeTillweDrain =
-                System.currentTimeMillis() + DRAIN_TIME_PERIOD;
+                System.currentTimeMillis() + drainTimeoutPeriod;
             while (!timelineEntityQueue.isEmpty()) {
               publishWithoutBlockingOnQueue(timelineEntityQueue.poll());
               if (System.currentTimeMillis() > timeTillweDrain) {
@@ -449,7 +452,7 @@ public class TimelineV2ClientImpl extends TimelineV2Client {
       LOG.info("Stopping TimelineClient.");
       executor.shutdownNow();
       try {
-        executor.awaitTermination(DRAIN_TIME_PERIOD, TimeUnit.MILLISECONDS);
+        executor.awaitTermination(drainTimeoutPeriod, TimeUnit.MILLISECONDS);
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         e.printStackTrace();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml

@@ -2101,6 +2101,15 @@
     <value>1000</value>
   </property>
 
+  <property>
+    <description>
+    The time period for which timeline v2 client will wait for draining
+    leftover entities after stop.
+    </description>
+    <name>yarn.timeline-service.client.drain-entities.timeout.ms</name>
+    <value>2000</value>
+  </property>
+
   <property>
     <description>Enable timeline server to recover state after starting. If
     true, then yarn.timeline-service.state-store-class must be specified.