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 1ebd43c2250 [FLINK-40099] Fix flaky YARNSessionFIFOSecuredITCase by
verifying Kerberos keytab logs before the application is killed
1ebd43c2250 is described below
commit 1ebd43c2250f82ef8bb03151927cb4d07104fc07
Author: Ramin Gharib <[email protected]>
AuthorDate: Wed Jul 8 13:19:45 2026 +0200
[FLINK-40099] Fix flaky YARNSessionFIFOSecuredITCase by verifying Kerberos
keytab logs before the application is killed
---
.../apache/flink/yarn/YARNSessionFIFOITCase.java | 22 +++++++++
.../flink/yarn/YARNSessionFIFOSecuredITCase.java | 55 +++++++++++++---------
2 files changed, 56 insertions(+), 21 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 9c895ca7b3c..db09d487365 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,6 +23,7 @@ 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;
@@ -95,6 +96,23 @@ 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");
@@ -178,6 +196,10 @@ 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 b4b972f03b7..6e92ae0a939 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
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.SecurityOptions;
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;
@@ -155,9 +156,12 @@ class YARNSessionFIFOSecuredITCase extends
YARNSessionFIFOITCase {
SecurityOptions.KERBEROS_LOGIN_PRINCIPAL.key(),
SecureTestEnvironment.getHadoopServicePrincipal());
}
- final ApplicationId applicationId =
- runDetachedModeTest(securityProperties, VIEW_ACLS,
MODIFY_ACLS);
- verifyResultContainsKerberosKeytab(applicationId,
VIEW_ACLS, MODIFY_ACLS);
+ runDetachedModeTest(
+ securityProperties,
+ VIEW_ACLS,
+ MODIFY_ACLS,
+
YARNSessionFIFOSecuredITCase::verifyKerberosKeytabInLogs);
+ verifyTokenAndAclsInContainers(VIEW_ACLS, MODIFY_ACLS);
});
}
@@ -177,29 +181,38 @@ class YARNSessionFIFOSecuredITCase extends
YARNSessionFIFOITCase {
SecurityOptions.KERBEROS_LOGIN_PRINCIPAL.key(),
SecureTestEnvironment.getHadoopServicePrincipal());
}
- 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);
+ 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);
});
}
- private static void verifyResultContainsKerberosKeytab(
- ApplicationId applicationId, String viewAcls, String modifyAcls)
throws Exception {
+ /**
+ * 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 {
final String[] mustHave = {"Login successful for user", "using keytab
file"};
- final boolean jobManagerRunsWithKerberos =
- verifyStringsInNamedLogFiles(mustHave, applicationId,
"jobmanager.log");
- final boolean taskManagerRunsWithKerberos =
- verifyStringsInNamedLogFiles(mustHave, applicationId,
"taskmanager.log");
-
- assertThat(jobManagerRunsWithKerberos &&
taskManagerRunsWithKerberos).isTrue();
+ 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);
+ }
+ }
+ 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");
