[incubator-gobblin] branch master updated: [GOBBLIN-1030] Refactor AbstractYarnSecurityManager to expose method for sending token file update message

[su...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

suvasude pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git


The following commit(s) were added to refs/heads/master by this push:
     new c7e2c67  [GOBBLIN-1030] Refactor AbstractYarnSecurityManager to expose method for sending token file update message
c7e2c67 is described below

commit c7e2c670734ea1009dace621ecb137219476ce29
Author: sv2000 <su...@gmail.com>
AuthorDate: Thu Jan 16 14:40:10 2020 -0800

    [GOBBLIN-1030] Refactor AbstractYarnSecurityManager to expose method for sending token file update message
    
    Closes #2872 from sv2000/helixClusterManagedConfig
---
 .../yarn/AbstractYarnAppSecurityManager.java       | 20 ++++++++++++++++-
 .../yarn/YarnAppSecurityManagerWithKeytabs.java    | 25 ++--------------------
 2 files changed, 21 insertions(+), 24 deletions(-)

diff --git a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/AbstractYarnAppSecurityManager.java b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/AbstractYarnAppSecurityManager.java
index af31fa2..d6da93e 100644
--- a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/AbstractYarnAppSecurityManager.java
+++ b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/AbstractYarnAppSecurityManager.java
@@ -44,6 +44,8 @@ import com.google.common.base.Throwables;
 import com.google.common.util.concurrent.AbstractIdleService;
 import com.typesafe.config.Config;
 
+import org.apache.gobblin.cluster.GobblinClusterConfigurationKeys;
+import org.apache.gobblin.cluster.GobblinClusterManager;
 import org.apache.gobblin.cluster.GobblinHelixMessagingService;
 import org.apache.gobblin.util.ConfigUtils;
 import org.apache.gobblin.util.ExecutorsUtils;
@@ -67,12 +69,15 @@ public abstract class AbstractYarnAppSecurityManager extends AbstractIdleService
   protected final HelixManager helixManager;
   protected final FileSystem fs;
   protected final Path tokenFilePath;
+
   protected Token<? extends TokenIdentifier> token;
   private final long loginIntervalInMinutes;
   private final long tokenRenewIntervalInMinutes;
-
+  private final boolean isHelixClusterManaged;
+  private final String helixInstanceName;
   private final ScheduledExecutorService loginExecutor;
   private final ScheduledExecutorService tokenRenewExecutor;
+
   private Optional<ScheduledFuture<?>> scheduledTokenRenewTask = Optional.absent();
 
   // This flag is used to tell if this is the first login. If yes, no token updated message will be
@@ -95,6 +100,10 @@ public abstract class AbstractYarnAppSecurityManager extends AbstractIdleService
         ExecutorsUtils.newThreadFactory(Optional.of(LOGGER), Optional.of("KeytabReLoginExecutor")));
     this.tokenRenewExecutor = Executors.newSingleThreadScheduledExecutor(
         ExecutorsUtils.newThreadFactory(Optional.of(LOGGER), Optional.of("TokenRenewExecutor")));
+    this.isHelixClusterManaged = ConfigUtils.getBoolean(config, GobblinClusterConfigurationKeys.IS_HELIX_CLUSTER_MANAGED,
+        GobblinClusterConfigurationKeys.DEFAULT_IS_HELIX_CLUSTER_MANAGED);
+    this.helixInstanceName = ConfigUtils.getString(config, GobblinClusterConfigurationKeys.HELIX_INSTANCE_NAME_KEY,
+        GobblinClusterManager.class.getSimpleName());
   }
 
   @Override
@@ -166,6 +175,15 @@ public abstract class AbstractYarnAppSecurityManager extends AbstractIdleService
     }
   }
 
+  protected void sendTokenFileUpdatedMessage() {
+    // Send a message to the controller (when the cluster is not managed)
+    // and all the participants if this is not the first login
+    if (!this.isHelixClusterManaged) {
+      sendTokenFileUpdatedMessage(InstanceType.CONTROLLER);
+    }
+    sendTokenFileUpdatedMessage(InstanceType.PARTICIPANT);
+  }
+
   /**
    * This method is used to send TokenFileUpdatedMessage which will handle by {@link YarnContainerSecurityManager}
    */
diff --git a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnAppSecurityManagerWithKeytabs.java b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnAppSecurityManagerWithKeytabs.java
index 5f234d2..8925d94 100644
--- a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnAppSecurityManagerWithKeytabs.java
+++ b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnAppSecurityManagerWithKeytabs.java
@@ -26,17 +26,12 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.helix.HelixManager;
-import org.apache.helix.InstanceType;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Optional;
 import com.google.common.base.Strings;
 import com.typesafe.config.Config;
 
-import org.apache.gobblin.cluster.GobblinClusterConfigurationKeys;
-import org.apache.gobblin.cluster.GobblinClusterManager;
-import org.apache.gobblin.util.ConfigUtils;
-
 
 /**
  * A class for managing Kerberos login and token renewing on the client side that has access to
@@ -61,8 +56,6 @@ import org.apache.gobblin.util.ConfigUtils;
  * @author Yinan Li
  */
 public class YarnAppSecurityManagerWithKeytabs extends AbstractYarnAppSecurityManager {
-
-  private final String helixInstanceName;
   private UserGroupInformation loginUser;
   private Optional<ScheduledFuture<?>> scheduledTokenRenewTask = Optional.absent();
 
@@ -70,16 +63,11 @@ public class YarnAppSecurityManagerWithKeytabs extends AbstractYarnAppSecurityMa
   // sent to the controller and the participants as they may not be up running yet. The first login
   // happens after this class starts up so the token gets regularly refreshed before the next login.
   private volatile boolean firstLogin = true;
-  private final boolean isHelixClusterManaged;
 
   public YarnAppSecurityManagerWithKeytabs(Config config, HelixManager helixManager, FileSystem fs, Path tokenFilePath)
       throws IOException {
     super(config, helixManager, fs, tokenFilePath);
     this.loginUser = UserGroupInformation.getLoginUser();
-    this.isHelixClusterManaged = ConfigUtils.getBoolean(config, GobblinClusterConfigurationKeys.IS_HELIX_CLUSTER_MANAGED,
-        GobblinClusterConfigurationKeys.DEFAULT_IS_HELIX_CLUSTER_MANAGED);
-    this.helixInstanceName = ConfigUtils.getString(config, GobblinClusterConfigurationKeys.HELIX_INSTANCE_NAME_KEY,
-        GobblinClusterManager.class.getSimpleName());
   }
 
   /**
@@ -90,12 +78,7 @@ public class YarnAppSecurityManagerWithKeytabs extends AbstractYarnAppSecurityMa
     writeDelegationTokenToFile();
 
     if (!this.firstLogin) {
-      // Send a message to the controller (when the cluster is not managed)
-      // and all the participants if this is not the first login
-      if (!this.isHelixClusterManaged) {
-        sendTokenFileUpdatedMessage(InstanceType.CONTROLLER);
-      }
-      sendTokenFileUpdatedMessage(InstanceType.PARTICIPANT, this.helixInstanceName);
+      sendTokenFileUpdatedMessage();
     }
   }
 
@@ -138,11 +121,7 @@ public class YarnAppSecurityManagerWithKeytabs extends AbstractYarnAppSecurityMa
     writeDelegationTokenToFile();
 
     if (!this.firstLogin) {
-      // Send a message to the controller (when the cluster is not managed) and all the participants
-      if (!this.isHelixClusterManaged) {
-        sendTokenFileUpdatedMessage(InstanceType.CONTROLLER);
-      }
-      sendTokenFileUpdatedMessage(InstanceType.PARTICIPANT, this.helixInstanceName);
+      sendTokenFileUpdatedMessage();
     }
   }
 }