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 c4e4b6bb672 [FLINK-40664][tests] Isolate MDC logging test clusters
(#29235)
c4e4b6bb672 is described below
commit c4e4b6bb672d58cce914856d0ed7d796a124503c
Author: Sankalp Thakur <[email protected]>
AuthorDate: Wed Oct 21 16:39:18 2026 +0400
[FLINK-40664][tests] Isolate MDC logging test clusters (#29235)
* [FLINK-40664][tests] Ignore Freeing inactive slots log in
JobIDLoggingITCase
Ignored log patterns now skip matching events even when they carry a
different MDC job ID. MiniCluster tests can observe TaskExecutor slot
cleanup for another job; that used to fail as "wrong value".
Generated-by: Cursor Grok 4.6
Co-authored-by: Sankalp Thakur <[email protected]>
* [FLINK-40664][tests] Isolate MDC logging clusters and preserve assertion
precedence
Generated-by: ChatGPT GPT-6 Astra Pro
---------
Co-authored-by: Sankalp Thakur <[email protected]>
---
.../apache/flink/util/JobIDLoggingUtilTest.java | 116 +++++++++++++++++++++
.../flink/test/misc/EnrichedMdcLoggingITCase.java | 99 ++++++++++++++++++
.../apache/flink/test/misc/JobIDLoggingITCase.java | 46 +-------
3 files changed, 217 insertions(+), 44 deletions(-)
diff --git
a/flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/util/JobIDLoggingUtilTest.java
b/flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/util/JobIDLoggingUtilTest.java
new file mode 100644
index 00000000000..4cb3a4ff85f
--- /dev/null
+++
b/flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/util/JobIDLoggingUtilTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.util;
+
+import org.apache.flink.testutils.logging.LoggerAuditingExtension;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+import org.slf4j.event.Level;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link JobIDLoggingUtil}. */
+class JobIDLoggingUtilTest {
+ private static final Logger LOG =
LoggerFactory.getLogger(JobIDLoggingUtilTest.class);
+ private static final String KEY = "flink-job-id";
+
+ @RegisterExtension
+ final LoggerAuditingExtension logging =
+ new LoggerAuditingExtension(JobIDLoggingUtilTest.class,
Level.DEBUG);
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "Freeing inactive slots for job job-b.",
+ "Free slot with allocation id allocation-b.",
+ "Close JobManager connection for job job-b.",
+ "Releasing job resources for job job-b."
+ })
+ void ignoredPatternsDoNotHideWrongJobIds(String message) {
+ logWithMdc("job-a", "Received task test-task.");
+ logWithMdc("job-b", message);
+ assertThatThrownBy(
+ () ->
+ JobIDLoggingUtil.assertKeyPresent(
+ KEY,
+ "job-a",
+ logging,
+ Collections.singletonList("Received
task .*"),
+ "Free.*|Close.*|Releasing.*"))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("events with a wrong value");
+ }
+
+ @Test
+ void expectedEventsStillCountWhenAnIgnorePatternMatches() {
+ logWithMdc("job-a", "Received task test-task.");
+ JobIDLoggingUtil.assertKeyPresent(
+ KEY,
+ "job-a",
+ logging,
+ Collections.singletonList("Received task .*"),
+ "Received task .*");
+ }
+
+ @Test
+ void eventsWithADifferentKeyValueFailWhenNotIgnored() {
+ logWithMdc("job-a", "Received task test-task.");
+ logWithMdc("job-b", "Freeing inactive slots for job job-b.");
+
+ assertThatThrownBy(
+ () ->
+ JobIDLoggingUtil.assertKeyPresent(
+ KEY,
+ "job-a",
+ logging,
+ Collections.singletonList("Received
task .*")))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("events with a wrong value");
+ }
+
+ @Test
+ void ignorePatternsStillSkipEventsWithAMissingKey() {
+ logWithMdc("job-a", "Received task test-task.");
+ LOG.debug("Successful registration at resource manager.");
+
+ JobIDLoggingUtil.assertKeyPresent(
+ KEY,
+ "job-a",
+ logging,
+ Collections.singletonList("Received task .*"),
+ "Successful registration.*");
+ }
+
+ private static void logWithMdc(String jobId, String message) {
+ MDC.put(KEY, jobId);
+ try {
+ LOG.debug(message);
+ } finally {
+ MDC.remove(KEY);
+ }
+ }
+}
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/misc/EnrichedMdcLoggingITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/misc/EnrichedMdcLoggingITCase.java
new file mode 100644
index 00000000000..7ad057f0fef
--- /dev/null
+++
b/flink-tests/src/test/java/org/apache/flink/test/misc/EnrichedMdcLoggingITCase.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MdcOptions;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.logging.LoggerAuditingExtension;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.apache.flink.test.misc.JobIDLoggingITCase.getConfiguration;
+import static org.apache.flink.test.misc.JobIDLoggingITCase.runJob;
+import static org.apache.flink.util.JobIDLoggingUtil.assertKeyPresent;
+import static org.slf4j.event.Level.DEBUG;
+
+class EnrichedMdcLoggingITCase {
+ @RegisterExtension
+ public final LoggerAuditingExtension jobMasterLogging =
+ new LoggerAuditingExtension(JobMaster.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension taskExecutorLogging =
+ new LoggerAuditingExtension(TaskExecutor.class, DEBUG);
+
+ @RegisterExtension
+ public static MiniClusterExtension miniClusterResource =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(getConfiguration())
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ void testEnrichedMdcLogging(@InjectClusterClient ClusterClient<?>
clusterClient)
+ throws Exception {
+ final Configuration enrichmentConfig = new Configuration();
+ enrichmentConfig.set(
+ MdcOptions.JOB_CONFIGURATION_TO_MDC_KEYS, Map.of("job.key-1",
"mdc-key-1"));
+ enrichmentConfig.setString("job.key-1", "val-1");
+
+ final JobID jobID = runJob(clusterClient, enrichmentConfig);
+ clusterClient.cancel(jobID).get();
+
+ assertKeyPresent(
+ "mdc-key-1",
+ "val-1",
+ jobMasterLogging,
+ asList("Initializing job .*", "Starting execution of job .*"),
+ "Registration at ResourceManager.*",
+ "Registration with ResourceManager.*",
+ "Resolved ResourceManager address.*");
+
+ assertKeyPresent(
+ "mdc-key-1",
+ "val-1",
+ taskExecutorLogging,
+ asList("Received task .*"),
+ "TaskManager received a checkpoint confirmation for unknown
task.*",
+ "TaskManager received an aborted checkpoint for unknown
task.*",
+ "Un-registering task.*",
+ "Successful registration.*",
+ "Establish JobManager connection.*",
+ "Offer reserved slots.*",
+ ".*ResourceManager.*",
+ "Operator event.*",
+ "Recovered slot allocation snapshots.*",
+ ".*heartbeat.*",
+ ".*leadership.*",
+ "Freeing inactive slots.*");
+ }
+}
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java
index aa9de906138..8f14634cee6 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java
@@ -27,7 +27,6 @@ import
org.apache.flink.api.connector.source.lib.NumberSequenceSource;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
-import org.apache.flink.configuration.MdcOptions;
import org.apache.flink.core.execution.CheckpointType;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
import org.apache.flink.runtime.checkpoint.CheckpointException;
@@ -51,7 +50,6 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
-import java.util.Map;
import java.util.concurrent.ExecutionException;
import static java.util.Arrays.asList;
@@ -107,7 +105,7 @@ class JobIDLoggingITCase {
.setNumberSlotsPerTaskManager(1)
.build());
- private static Configuration getConfiguration() {
+ static Configuration getConfiguration() {
Configuration configuration = new Configuration();
configuration.set(SCHEDULER, JobManagerOptions.SchedulerType.Adaptive);
return configuration;
@@ -230,51 +228,11 @@ class JobIDLoggingITCase {
".* finished asynchronous part of checkpoint .*"));
}
- @Test
- void testEnrichedMdcLogging(@InjectClusterClient ClusterClient<?>
clusterClient)
- throws Exception {
- final Configuration enrichmentConfig = new Configuration();
- enrichmentConfig.set(
- MdcOptions.JOB_CONFIGURATION_TO_MDC_KEYS, Map.of("job.key-1",
"mdc-key-1"));
- enrichmentConfig.setString("job.key-1", "val-1");
-
- final JobID jobID = runJob(clusterClient, enrichmentConfig);
- clusterClient.cancel(jobID).get();
-
- assertKeyPresent(
- "mdc-key-1",
- "val-1",
- jobMasterLogging,
- asList("Initializing job .*", "Starting execution of job .*"),
- "Registration at ResourceManager.*",
- "Registration with ResourceManager.*",
- "Resolved ResourceManager address.*");
-
- assertKeyPresent(
- "mdc-key-1",
- "val-1",
- taskExecutorLogging,
- asList("Received task .*"),
- "TaskManager received a checkpoint confirmation for unknown
task.*",
- "TaskManager received an aborted checkpoint for unknown
task.*",
- "Un-registering task.*",
- "Successful registration.*",
- "Establish JobManager connection.*",
- "Offer reserved slots.*",
- ".*ResourceManager.*",
- "Operator event.*",
- "Recovered slot allocation snapshots.*",
- ".*heartbeat.*",
- ".*leadership.*",
- "Freeing inactive slots.*");
- }
-
private static JobID runJob(ClusterClient<?> clusterClient) throws
Exception {
return runJob(clusterClient, new Configuration());
}
- private static JobID runJob(ClusterClient<?> clusterClient, Configuration
jobConfig)
- throws Exception {
+ static JobID runJob(ClusterClient<?> clusterClient, Configuration
jobConfig) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.fromSource(
