[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)

Author: MartijnVisser

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java

@@ -23,7 +23,6 @@
 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 @@ void testDetachedMode() throws Exception {
     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 @@ ApplicationId runDetachedModeTest(
                                 "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();

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOSecuredITCase.java

@@ -20,10 +20,10 @@
 
 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.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 @@ void testDetachedModeSecureWithPreInstallKeytab() throws Exception {
                                 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 @@ void testDetachedMode() throws Exception {
                                 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");