diff --git a/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java b/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java index 59e7a584af0..f4c24b03215 100644 --- a/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java +++ b/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) diff --git a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java index e4acb8682eb..ee0fc33683d 100644 --- a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java +++ b/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. *

* + *

+ * Users of {@link GobblinYarnAppLauncher} need to call {@link #initializeYarnClients} which a child class can override. + *

+ * * @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 potentialYarnClients) { + public void initializeYarnClients(Config config) { Set 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); } }