initiliaze yarn clients in yarn app launcher so that a child class ca…

…n override the yarn client creation logic

gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java

@@ -76,6 +76,7 @@ public AzkabanGobblinYarnAppLauncher(String jobId, Properties gobblinProps)
     gobblinConfig = gobblinConfig.withValue(GobblinYarnAppLauncher.GOBBLIN_YARN_APP_LAUNCHER_MODE,
         ConfigValueFactory.fromAnyRef(GobblinYarnAppLauncher.AZKABAN_APP_LAUNCHER_MODE_KEY));
     this.gobblinYarnAppLauncher = getYarnAppLauncher(gobblinConfig);
+    this.gobblinYarnAppLauncher.initializeYarnClients(gobblinConfig);
   }
 
   protected GobblinYarnAppLauncher getYarnAppLauncher(Config gobblinConfig)

gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java

@@ -157,6 +157,10 @@
  *   this count exceeds the maximum number allowed, it will initiate a shutdown.
  * </p>
  *
+ * <p>
+ *   Users of {@link GobblinYarnAppLauncher} need to call {@link #initializeYarnClients} which a child class can override.
+ * </p>
+ *
  * @author Yinan Li
  */
 public class GobblinYarnAppLauncher {
@@ -260,7 +264,6 @@ public GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration
     YarnHelixUtils.setAdditionalYarnClassPath(config, this.yarnConfiguration);
     this.yarnConfiguration.set("fs.automatic.close", "false");
     this.originalYarnRMAddress = this.yarnConfiguration.get(GobblinYarnConfigurationKeys.YARN_RESOURCE_MANAGER_ADDRESS);
-    createPotentialYarnClients(config, this.potentialYarnClients);
 
     this.fs = GobblinClusterUtils.buildFileSystem(config, this.yarnConfiguration);
     this.closer.register(this.fs);
@@ -323,14 +326,14 @@ public GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration
     }
   }
 
-  protected void createPotentialYarnClients(Config config, Map<String, YarnClient> potentialYarnClients) {
+  public void initializeYarnClients(Config config) {
     Set<String> potentialRMAddresses = new HashSet<>(ConfigUtils.getStringList(config, GobblinYarnConfigurationKeys.OTHER_YARN_RESOURCE_MANAGER_ADDRESSES));
     potentialRMAddresses.add(originalYarnRMAddress);
     for (String rmAddress : potentialRMAddresses) {
       YarnClient tmpYarnClient = YarnClient.createYarnClient();
       this.yarnConfiguration.set(GobblinYarnConfigurationKeys.YARN_RESOURCE_MANAGER_ADDRESS, rmAddress);
       tmpYarnClient.init(new YarnConfiguration(this.yarnConfiguration));
-      potentialYarnClients.put(rmAddress, tmpYarnClient);
+      this.potentialYarnClients.put(rmAddress, tmpYarnClient);
     }
   }