YARN-9425. Make initialDelay configurable for FederationStateStoreService#scheduledExecutorService

Author: Ashutosh Gupta

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

@@ -3920,6 +3920,13 @@ public static boolean isAclEnabled(Configuration conf) {
   public static final String DEFAULT_FEDERATION_REGISTRY_BASE_KEY =
       "yarnfederation/";
 
+  public static final String FEDERATION_STATESTORE_HEARTBEAT_INITIAL_DELAY_SECS =
+      FEDERATION_PREFIX + "state-store.heartbeat.initial-delay-secs";
+
+  // 30 secs
+  public static final int
+      DEFAULT_FEDERATION_STATESTORE_HEARTBEAT_INITIAL_DELAY_SECS = 30;
+
   public static final String FEDERATION_STATESTORE_HEARTBEAT_INTERVAL_SECS =
       FEDERATION_PREFIX + "state-store.heartbeat-interval-secs";
 

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

@@ -3624,6 +3624,13 @@
     <name>yarn.federation.enabled</name>
     <value>false</value>
   </property>
+  <property>
+    <description>
+      Initial delay for federation state-store heartbeat service.
+    </description>
+    <name>yarn.federation.state-store.heartbeat.initial-delay-secs</name>
+    <value>30</value>
+  </property>
   <property>
     <description>
       Machine list file to be loaded by the FederationSubCluster Resolver

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/federation/FederationStateStoreService.java

@@ -86,6 +86,7 @@ public class FederationStateStoreService extends AbstractService
   private FederationStateStore stateStoreClient = null;
   private SubClusterId subClusterId;
   private long heartbeatInterval;
+  private long heartbeatInitialDelay;
   private RMContext rmContext;
 
   public FederationStateStoreService(RMContext rmContext) {
@@ -120,6 +121,14 @@ protected void serviceInit(Configuration conf) throws Exception {
       heartbeatInterval =
           YarnConfiguration.DEFAULT_FEDERATION_STATESTORE_HEARTBEAT_INTERVAL_SECS;
     }
+
+    heartbeatInitialDelay = conf.getLong(
+        YarnConfiguration.FEDERATION_STATESTORE_HEARTBEAT_INITIAL_DELAY_SECS,
+        YarnConfiguration.DEFAULT_FEDERATION_STATESTORE_HEARTBEAT_INITIAL_DELAY_SECS);
+    if (heartbeatInitialDelay <= 0) {
+      heartbeatInitialDelay =
+          YarnConfiguration.DEFAULT_FEDERATION_STATESTORE_HEARTBEAT_INITIAL_DELAY_SECS;
+    }
     LOG.info("Initialized federation membership service.");
 
     super.serviceInit(conf);
@@ -196,7 +205,7 @@ private void registerAndInitializeHeartbeat() {
     scheduledExecutorService =
         HadoopExecutors.newSingleThreadScheduledExecutor();
     scheduledExecutorService.scheduleWithFixedDelay(stateStoreHeartbeat,
-        heartbeatInterval, heartbeatInterval, TimeUnit.SECONDS);
+        heartbeatInitialDelay, heartbeatInterval, TimeUnit.SECONDS);
     LOG.info("Started federation membership heartbeat with interval: {}",
         heartbeatInterval);
   }