[FLINK-28608][runtime][security]Make Hadoop FS token renewer configur…

…able
zhuoluoy · Jul 21, 2022 · 0c5108c · 0c5108c
1 parent 5f0c4ab
commit 0c5108c
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 1 deletion.
diff --git a/...rc/main/java/org/apache/flink/runtime/security/token/HadoopFSDelegationTokenProvider.java b/...rc/main/java/org/apache/flink/runtime/security/token/HadoopFSDelegationTokenProvider.java
@@ -78,7 +78,7 @@ public Optional<Long> obtainDelegationTokens(Credentials credentials) throws Exc
         Clock clock = Clock.systemDefaultZone();
         Set<FileSystem> fileSystemsToAccess = getFileSystemsToAccess();
 
-        obtainDelegationTokens(null, fileSystemsToAccess, credentials);
+        obtainDelegationTokens(getRenewer(), fileSystemsToAccess, credentials);
 
         // Get the token renewal interval if it is not set. It will be called only once.
         if (tokenRenewalInterval == null) {
@@ -88,6 +88,13 @@ public Optional<Long> obtainDelegationTokens(Credentials credentials) throws Exc
                 interval -> getTokenRenewalDate(clock, credentials, interval));
     }
 
+    @VisibleForTesting
+    @Nullable
+    String getRenewer() {
+        return flinkConfiguration.getString(
+                String.format("security.kerberos.token.provider.%s.renewer", serviceName()), null);
+    }
+
     private Set<FileSystem> getFileSystemsToAccess() throws IOException {
         Set<FileSystem> result = new HashSet<>();
 
@@ -155,6 +162,7 @@ protected void obtainDelegationTokens(
                 });
     }
 
+    @VisibleForTesting
     Optional<Long> getTokenRenewalInterval(Clock clock, Set<FileSystem> fileSystemsToAccess)
             throws IOException {
         // We cannot use the tokens generated with renewer yarn

diff --git a/...t/java/org/apache/flink/runtime/security/token/HadoopFSDelegationTokenProviderITCase.java b/...t/java/org/apache/flink/runtime/security/token/HadoopFSDelegationTokenProviderITCase.java
@@ -35,6 +35,7 @@
 
 import static java.time.Instant.ofEpochMilli;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 /** Test for {@link HadoopFSDelegationTokenProvider}. */
 class HadoopFSDelegationTokenProviderITCase {
@@ -66,6 +67,24 @@ public long renew(Configuration conf) {
         }
     }
 
+    @Test
+    public void getRenewerShouldReturnNullByDefault() throws Exception {
+        HadoopFSDelegationTokenProvider provider = new HadoopFSDelegationTokenProvider();
+        provider.init(new org.apache.flink.configuration.Configuration());
+        assertNull(provider.getRenewer());
+    }
+
+    @Test
+    public void getRenewerShouldReturnConfiguredRenewer() throws Exception {
+        String renewer = "testRenewer";
+        HadoopFSDelegationTokenProvider provider = new HadoopFSDelegationTokenProvider();
+        org.apache.flink.configuration.Configuration configuration =
+                new org.apache.flink.configuration.Configuration();
+        configuration.setString("security.kerberos.token.provider.hadoopfs.renewer", renewer);
+        provider.init(configuration);
+        assertEquals(renewer, provider.getRenewer());
+    }
+
     @Test
     public void getTokenRenewalIntervalShouldReturnNoneWhenNoTokens() throws IOException {
         HadoopFSDelegationTokenProvider provider =