This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b464b331dbd [FLINK-40109] Verify Kerberos keytab logs after the YARN
application is killed
b464b331dbd is described below
commit b464b331dbd2cbd4714c47816cb0558b14955c91
Author: Martijn Visser <[email protected]>
AuthorDate: Thu Jul 9 12:49:36 2026 +0200
[FLINK-40109] Verify Kerberos keytab logs after the YARN application is
killed
FLINK-40099 moved YARNSessionFIFOSecuredITCase's keytab check into a
pre-kill
callback polling waitUntilCondition(cond, 500). The 500 int literal binds to
the (cond, int retryAttempts) overload (500*100ms, then "Exhausted retry
attempts."), and reading the JobManager log while the application is still
alive is racy: the polled jobmanager.log does not reliably contain the
keytab
login line before teardown.
Read the container logs after the application is killed (output flushed) but
poll each log with a bounded wait instead of a single read, preserving the
FLINK-17662 intent without the pre-kill race. On timeout, fail with the log
file and expected strings so a recurrence is diagnosable from the uploaded
container logs.
Generated-by: Claude Opus 4.8 (1M context)
---
.../apache/flink/yarn/YARNSessionFIFOITCase.java | 22 --------
.../flink/yarn/YARNSessionFIFOSecuredITCase.java | 63 +++++++++++-----------
2 files changed, 32 insertions(+), 53 deletions(-)
diff --git
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
index db09d487365..9c895ca7b3c 100644
---
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
+++
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
@@ -23,7 +23,6 @@ import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.test.testdata.WordCountData;
import org.apache.flink.testutils.logging.LoggerAuditingExtension;
-import org.apache.flink.util.function.ThrowingConsumer;
import org.apache.flink.yarn.cli.FlinkYarnSessionCli;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
@@ -96,23 +95,6 @@ class YARNSessionFIFOITCase extends YarnTestBase {
ApplicationId runDetachedModeTest(
Map<String, String> securityProperties, String viewAcls, String
modifyAcls)
throws Exception {
- return runDetachedModeTest(securityProperties, viewAcls, modifyAcls,
ignored -> {});
- }
-
- /**
- * Test regular operation, including command line parameter parsing.
- *
- * @param verifyWhileRunning verification that is invoked once the job has
finished but while
- * the YARN application and its containers are still alive, before the
application is
- * killed. This allows subclasses to inspect complete container logs
and live container
- * state, which is no longer possible once the application has been
torn down.
- */
- ApplicationId runDetachedModeTest(
- Map<String, String> securityProperties,
- String viewAcls,
- String modifyAcls,
- ThrowingConsumer<ApplicationId, Exception> verifyWhileRunning)
- throws Exception {
log.info("Starting testDetachedMode()");
File exampleJarLocation = getTestJarPath("StreamingWordCount.jar");
@@ -196,10 +178,6 @@ class YARNSessionFIFOITCase extends YarnTestBase {
"jobmanager.log"),
testConditionIntervalInMillis);
- // Run verification that requires the application and its containers
to still be alive, such
- // as reading the freshly written container logs, before the
application is killed below.
- verifyWhileRunning.accept(applicationId);
-
// kill application "externally".
try {
YarnClient yc = YarnClient.createYarnClient();
diff --git
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOSecuredITCase.java
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOSecuredITCase.java
index 6e92ae0a939..3ef2992ec93 100644
---
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOSecuredITCase.java
+++
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOSecuredITCase.java
@@ -20,10 +20,10 @@ package org.apache.flink.yarn;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.SecurityOptions;
+import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.security.SecurityConfiguration;
import org.apache.flink.runtime.security.SecurityUtils;
import org.apache.flink.runtime.security.contexts.HadoopSecurityContext;
-import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.test.util.SecureTestEnvironment;
import org.apache.flink.test.util.TestingSecurityContext;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
@@ -47,6 +47,8 @@ import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Duration;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -156,12 +158,9 @@ class YARNSessionFIFOSecuredITCase extends
YARNSessionFIFOITCase {
SecurityOptions.KERBEROS_LOGIN_PRINCIPAL.key(),
SecureTestEnvironment.getHadoopServicePrincipal());
}
- runDetachedModeTest(
- securityProperties,
- VIEW_ACLS,
- MODIFY_ACLS,
-
YARNSessionFIFOSecuredITCase::verifyKerberosKeytabInLogs);
- verifyTokenAndAclsInContainers(VIEW_ACLS, MODIFY_ACLS);
+ final ApplicationId applicationId =
+ runDetachedModeTest(securityProperties, VIEW_ACLS,
MODIFY_ACLS);
+ verifyResultContainsKerberosKeytab(applicationId,
VIEW_ACLS, MODIFY_ACLS);
});
}
@@ -181,38 +180,40 @@ class YARNSessionFIFOSecuredITCase extends
YARNSessionFIFOITCase {
SecurityOptions.KERBEROS_LOGIN_PRINCIPAL.key(),
SecureTestEnvironment.getHadoopServicePrincipal());
}
- runDetachedModeTest(
- securityProperties,
- VIEW_ACLS,
- MODIFY_ACLS,
-
YARNSessionFIFOSecuredITCase::verifyKerberosKeytabInLogs);
- verifyTokenAndAclsInContainers(VIEW_ACLS, MODIFY_ACLS);
- runDetachedModeTest(
- securityProperties,
- VIEW_ACLS_WITH_WILDCARD,
- MODIFY_ACLS_WITH_WILDCARD,
-
YARNSessionFIFOSecuredITCase::verifyKerberosKeytabInLogs);
- verifyTokenAndAclsInContainers(WILDCARD, WILDCARD);
+ final ApplicationId applicationId =
+ runDetachedModeTest(securityProperties, VIEW_ACLS,
MODIFY_ACLS);
+ verifyResultContainsKerberosKeytab(applicationId,
VIEW_ACLS, MODIFY_ACLS);
+ final ApplicationId applicationIdWithWildcard =
+ runDetachedModeTest(
+ securityProperties,
+ VIEW_ACLS_WITH_WILDCARD,
+ MODIFY_ACLS_WITH_WILDCARD);
+ verifyResultContainsKerberosKeytab(
+ applicationIdWithWildcard, WILDCARD, WILDCARD);
});
}
- /**
- * Verifies that both the JobManager and TaskManager logged a Kerberos
keytab login. The
- * container log files are polled rather than read once, and this must run
while the containers
- * are still alive so the logs are complete. A short-lived TaskManager can
otherwise be torn
- * down before its startup log becomes durably readable.
- */
- private static void verifyKerberosKeytabInLogs(ApplicationId
applicationId) throws Exception {
+ private static void verifyResultContainsKerberosKeytab(
+ ApplicationId applicationId, String viewAcls, String modifyAcls)
throws Exception {
final String[] mustHave = {"Login successful for user", "using keytab
file"};
+ // The application has been killed by now, so all container output is
flushed. A
+ // short-lived TaskManager's startup login line can still be briefly
unreadable right
+ // after teardown (FLINK-17662), so poll each log instead of reading
it once.
for (final String logFile : new String[] {"jobmanager.log",
"taskmanager.log"}) {
log.info("Waiting until {} contains the Kerberos keytab login",
logFile);
- CommonTestUtils.waitUntilCondition(
- () -> verifyStringsInNamedLogFiles(mustHave,
applicationId, logFile), 500);
+ CommonTestUtils.waitUtil(
+ () -> verifyStringsInNamedLogFiles(mustHave,
applicationId, logFile),
+ Duration.ofSeconds(60),
+ Duration.ofMillis(500),
+ "Kerberos keytab login "
+ + Arrays.toString(mustHave)
+ + " not found in "
+ + logFile
+ + " for application "
+ + applicationId
+ + " within the timeout; inspect the uploaded
container logs.");
}
- }
- private static void verifyTokenAndAclsInContainers(String viewAcls, String
modifyAcls)
- throws Exception {
final List<String> amRMTokens =
Lists.newArrayList(AMRMTokenIdentifier.KIND_NAME.toString());
final String jobmanagerContainerId =
getContainerIdByLogName("jobmanager.log");
