KAFKA-17767 Automatically quarantine new tests [5/n] (#17725)

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

.github/actions/run-gradle/action.yml

@@ -0,0 +1,75 @@
+# 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.
+#
+---
+name: "Gradle Setup"
+description: "Setup Java and Gradle"
+inputs:
+  # Composite actions do not support typed parameters. Everything is treated as a string
+  # See: https://github.com/actions/runner/issues/2238
+  test-task:
+    description: "The test suite to run. Either 'test' or 'quarantinedTest'."
+    required: true
+  timeout-minutes:
+    description: "The timeout for the tests, in minutes."
+    required: true
+  test-catalog-path:
+    description: "The file path of the test catalog file."
+    required: true
+  build-scan-artifact-name:
+    description: "The name to use for archiving the build scan."
+    required: true
+outputs:
+  gradle-exitcode:
+    description: "The result of the Gradle test task."
+    value: ${{ steps.run-tests.outputs.exitcode }}
+runs:
+  using: "composite"
+  steps:
+    - name: Run JUnit Tests (${{ inputs.test-task }})
+      # Gradle flags
+      # --build-cache:  Let Gradle restore the build cache
+      # --no-scan:      Don't attempt to publish the scan yet. We want to archive it first.
+      # --continue:     Keep running even if a test fails
+      # -PcommitId      Prevent the Git SHA being written into the jar files (which breaks caching)
+      shell: bash
+      id: run-tests
+      env:
+        TIMEOUT_MINUTES: ${{ inputs.timeout-minutes}}
+        TEST_CATALOG: ${{ inputs.test-catalog-path }}
+        TEST_TASK: ${{ inputs.test-task }}
+      run: |
+        set +e
+        ./.github/scripts/thread-dump.sh &
+        timeout ${TIMEOUT_MINUTES}m ./gradlew --build-cache --continue --no-scan \
+        -PtestLoggingEvents=started,passed,skipped,failed \
+        -PmaxParallelForks=2 \
+        -PmaxTestRetries=1 -PmaxTestRetryFailures=3 \
+        -PmaxQuarantineTestRetries=3 -PmaxQuarantineTestRetryFailures=0 \
+        -Pkafka.test.catalog.file=$TEST_CATALOG \
+        -PcommitId=xxxxxxxxxxxxxxxx \
+        $TEST_TASK
+        exitcode="$?"
+        echo "exitcode=$exitcode" >> $GITHUB_OUTPUT
+    - name: Archive build scan (${{ inputs.test-task }})
+      if: always()
+      uses: actions/upload-artifact@v4
+      with:
+        name: ${{ inputs.build-scan-artifact-name }}
+        path: ~/.gradle/build-scan-data
+        compression-level: 9
+        if-no-files-found: ignore

.github/scripts/junit.py

@@ -142,6 +142,7 @@ def parse_report(workspace_path, report_path, fp) -> Iterable[TestSuite]:
     cur_suite: Optional[TestSuite] = None
     partial_test_case = None
     test_case_failed = False
+    test_case_skipped = False
     for (event, elem) in xml.etree.ElementTree.iterparse(fp, events=["start", "end"]):
         if event == "start":
             if elem.tag == "testsuite":
@@ -171,11 +172,12 @@ def parse_report(workspace_path, report_path, fp) -> Iterable[TestSuite]:
             elif elem.tag == "skipped":
                 skipped = partial_test_case(None, None, None)
                 cur_suite.skipped_tests.append(skipped)
+                test_case_skipped = True
             else:
                 pass
         elif event == "end":
             if elem.tag == "testcase":
-                if not test_case_failed:
+                if not test_case_failed and not test_case_skipped:
                     passed = partial_test_case(None, None, None)
                     cur_suite.passed_tests.append(passed)
                 partial_test_case = None
@@ -303,7 +305,7 @@ if __name__ == "__main__":
                     logger.debug(f"Found skipped test: {skipped_test}")
                     skipped_table.append((simple_class_name, skipped_test.test_name))
 
-                # Collect all tests that were run as part of quarantinedTest
+                # Only collect quarantined tests from the "quarantinedTest" task
                 if task == "quarantinedTest":
                     for test in all_suite_passed.values():
                         simple_class_name = test.class_name.split(".")[-1]
@@ -329,53 +331,75 @@ if __name__ == "__main__":
     # The stdout (print) goes to the workflow step console output.
     # The stderr (logger) is redirected to GITHUB_STEP_SUMMARY which becomes part of the HTML job summary.
     report_url = get_env("JUNIT_REPORT_URL")
+    if report_url:
         report_md = f"Download [HTML report]({report_url})."
-    summary = (f"{total_run} tests cases run in {duration}. "
+    else:
+        report_md = "No report available. JUNIT_REPORT_URL was missing."
+    summary = (f"{total_run} tests cases run in {duration}.\n\n"
                f"{total_success} {PASSED}, {total_failures} {FAILED}, "
-               f"{total_flaky} {FLAKY}, {total_skipped} {SKIPPED}, and {total_errors} errors.")
+               f"{total_flaky} {FLAKY}, {total_skipped} {SKIPPED}, {len(quarantined_table)} {QUARANTINED}, and {total_errors} errors.<br/>")
     print("## Test Summary\n")
-    print(f"{summary} {report_md}\n")
+    print(f"{summary}\n\n{report_md}\n")
+
+    # Failed
     if len(failed_table) > 0:
-        logger.info(f"Found {len(failed_table)} test failures:")
+        print("<details open=\"true\">")
-        print("### Failed Tests\n")
+        print(f"<summary>Failed Tests {FAILED} ({len(failed_table)})</summary>\n")
         print(f"| Module | Test | Message | Time |")
         print(f"| ------ | ---- | ------- | ---- |")
+        logger.info(f"Found {len(failed_table)} test failures:")
         for row in failed_table:
             logger.info(f"{FAILED} {row[0]} > {row[1]}")
             row_joined = " | ".join(row)
             print(f"| {row_joined} |")
+        print("\n</details>")
     print("\n")
+
+    # Flaky
     if len(flaky_table) > 0:
-        logger.info(f"Found {len(flaky_table)} flaky test failures:")
+        print("<details open=\"true\">")
-        print("### Flaky Tests\n")
+        print(f"<summary>Flaky Tests {FLAKY} ({len(flaky_table)})</summary>\n")
         print(f"| Module | Test | Message | Time |")
         print(f"| ------ | ---- | ------- | ---- |")
+        logger.info(f"Found {len(flaky_table)} flaky test failures:")
         for row in flaky_table:
             logger.info(f"{FLAKY} {row[0]} > {row[1]}")
             row_joined = " | ".join(row)
             print(f"| {row_joined} |")
+        print("\n</details>")
     print("\n")
+
+    # Skipped
     if len(skipped_table) > 0:
         print("<details>")
-        print(f"<summary>{len(skipped_table)} Skipped Tests</summary>\n")
+        print(f"<summary>Skipped Tests {SKIPPED} ({len(skipped_table)})</summary>\n")
         print(f"| Module | Test |")
         print(f"| ------ | ---- |")
+        logger.debug(f"::group::Found {len(skipped_table)} skipped tests")
         for row in skipped_table:
             row_joined = " | ".join(row)
             print(f"| {row_joined} |")
+            logger.debug(f"{row[0]} > {row[1]}")
         print("\n</details>")
+        logger.debug("::endgroup::")
+    print("\n")
 
+    # Quarantined
     if len(quarantined_table) > 0:
-        logger.info(f"Ran {len(quarantined_table)} quarantined test:")
         print("<details>")
-        print(f"<summary>{len(quarantined_table)} Quarantined Tests</summary>\n")
+        print(f"<summary>Quarantined Tests {QUARANTINED} ({len(quarantined_table)})</summary>\n")
         print(f"| Module | Test |")
         print(f"| ------ | ---- |")
+        logger.debug(f"::group::Found {len(quarantined_table)} quarantined tests")
         for row in quarantined_table:
-            logger.info(f"{QUARANTINED} {row[0]} > {row[1]}")
             row_joined = " | ".join(row)
             print(f"| {row_joined} |")
+            logger.debug(f"{row[0]} > {row[1]}")
         print("\n</details>")
+        logger.debug("::endgroup::")
+
+    # Create a horizontal rule
+    print("-"*80)
 
     # Print special message if there was a timeout
     exit_code = get_env("GRADLE_EXIT_CODE", int)

.github/workflows/build.yml

@@ -165,32 +165,30 @@ jobs:
       # If the load-catalog job failed, we won't be able to download the artifact. Since we don't want this to fail
       # the overall workflow, so we'll continue here without a test catalog.
       - name: Load Test Catalog
+        id: load-test-catalog
         uses: actions/download-artifact@v4
         continue-on-error: true
         with:
           name: combined-test-catalog
 
-      - name: Test
+      - name: JUnit Quarantined Tests
-        # Gradle flags
+        id: junit-quarantined-test
-        # --build-cache:  Let Gradle restore the build cache
+        uses: ./.github/actions/run-gradle
-        # --no-scan:      Don't attempt to publish the scan yet. We want to archive it first.
+        with:
-        # --continue:     Keep running even if a test fails
+          test-task: quarantinedTest
-        # -PcommitId      Prevent the Git SHA being written into the jar files (which breaks caching)
+          timeout-minutes: 30
+          test-catalog-path: ${{ steps.load-test-catalog.outputs.download-path }}/combined-test-catalog.txt
+          build-scan-artifact-name: build-scan-quarantined-test-${{ matrix.java }}
+
+      - name: JUnit Tests
         id: junit-test
-        env:
+        uses: ./.github/actions/run-gradle
-          TIMEOUT_MINUTES: 180  # 3 hours
+        with:
-        run: |
+          test-task: test
-          set +e
+          timeout-minutes: 180  # 3 hours
-          ./.github/scripts/thread-dump.sh &
+          test-catalog-path: ${{ steps.load-test-catalog.outputs.download-path }}/combined-test-catalog.txt
-          timeout ${TIMEOUT_MINUTES}m ./gradlew --build-cache --continue --no-scan \
+          build-scan-artifact-name: build-scan-test-${{ matrix.java }}
-          -PtestLoggingEvents=started,passed,skipped,failed \
+
-          -PmaxParallelForks=2 \
-          -PmaxTestRetries=1 -PmaxTestRetryFailures=3 \
-          -PmaxQuarantineTestRetries=3 -PmaxQuarantineTestRetryFailures=0 \
-          -PcommitId=xxxxxxxxxxxxxxxx \
-          quarantinedTest test
-          exitcode="$?"
-          echo "exitcode=$exitcode" >> $GITHUB_OUTPUT
       - name: Archive JUnit HTML reports
         uses: actions/upload-artifact@v4
         id: junit-upload-artifact
@@ -200,6 +198,7 @@ jobs:
             **/build/reports/tests/*
           compression-level: 9
           if-no-files-found: ignore
+
       - name: Archive JUnit XML
         uses: actions/upload-artifact@v4
         with:
@@ -208,9 +207,10 @@ jobs:
             build/junit-xml/**/*.xml
           compression-level: 9
           if-no-files-found: ignore
+
       - name: Archive Thread Dumps
         id: thread-dump-upload-artifact
-        if: always() && steps.junit-test.outputs.exitcode == '124'
+        if: always() && (steps.junit-test.outputs.gradle-exitcode == '124' || steps.junit-quarantined-test.outputs.gradle-exitcode == '124')
         uses: actions/upload-artifact@v4
         with:
           name: junit-thread-dumps-${{ matrix.java }}
@@ -218,13 +218,15 @@ jobs:
             thread-dumps/*
           compression-level: 9
           if-no-files-found: ignore
+
       - name: Parse JUnit tests
         run: python .github/scripts/junit.py --export-test-catalog ./test-catalog >> $GITHUB_STEP_SUMMARY
         env:
           GITHUB_WORKSPACE: ${{ github.workspace }}
           JUNIT_REPORT_URL: ${{ steps.junit-upload-artifact.outputs.artifact-url }}
           THREAD_DUMP_URL: ${{ steps.thread-dump-upload-artifact.outputs.artifact-url }}
-          GRADLE_EXIT_CODE: ${{ steps.junit-test.outputs.exitcode }}
+          GRADLE_EXIT_CODE: ${{ steps.junit-test.outputs.gradle-exitcode }}
+
       - name: Archive Test Catalog
         if: ${{ always() && matrix.java == '23' }}
         uses: actions/upload-artifact@v4
@@ -233,14 +235,6 @@ jobs:
           path: test-catalog
           compression-level: 9
           if-no-files-found: ignore
-      - name: Archive Build Scan
-        if: always()
-        uses: actions/upload-artifact@v4
-        with:
-          name: build-scan-test-${{ matrix.java }}
-          path: ~/.gradle/build-scan-data
-          compression-level: 9
-          if-no-files-found: ignore
 
   update-test-catalog:
     name: Update Test Catalog

.github/workflows/ci-complete.yml

@@ -44,6 +44,7 @@ jobs:
       fail-fast: false
       matrix:
         java: [ 23, 11 ]
+        artifact-prefix: [ "build-scan-test-", "build-scan-quarantined-test-"]
     steps:
       - name: Env
         run: printenv
@@ -66,7 +67,7 @@ jobs:
         with:
           github-token: ${{ github.token }}
           run-id: ${{ github.event.workflow_run.id }}
-          name: build-scan-test-${{ matrix.java }}
+          name: ${{ matrix.artifact-prefix }}-${{ matrix.java }}
           path: ~/.gradle/build-scan-data  # This is where Gradle buffers unpublished build scan data when --no-scan is given
       - name: Handle missing scan
         if: ${{ steps.download-build-scan.outcome == 'failure' }}

build.gradle

@@ -135,6 +135,7 @@ ext {
   runtimeTestLibs = [
     libs.slf4jReload4j,
     libs.junitPlatformLanucher,
+    project(":test-common:test-common-runtime")
   ]
 }
 
@@ -483,6 +484,8 @@ subprojects {
 
     // KAFKA-17433 Used by deflake.yml github action to repeat individual tests
     systemProperty("kafka.cluster.test.repeat", project.findProperty("kafka.cluster.test.repeat"))
+    systemProperty("kafka.test.catalog.file", project.findProperty("kafka.test.catalog.file"))
+    systemProperty("kafka.test.run.quarantined", "false")
 
     testLogging {
       events = userTestLoggingEvents ?: testLoggingEvents
@@ -553,6 +556,8 @@ subprojects {
 
     // KAFKA-17433 Used by deflake.yml github action to repeat individual tests
     systemProperty("kafka.cluster.test.repeat", project.findProperty("kafka.cluster.test.repeat"))
+    systemProperty("kafka.test.catalog.file", project.findProperty("kafka.test.catalog.file"))
+    systemProperty("kafka.test.run.quarantined", "true")
 
     testLogging {
       events = userTestLoggingEvents ?: testLoggingEvents
@@ -564,7 +569,6 @@ subprojects {
 
     useJUnitPlatform {
       includeEngines 'junit-jupiter'
-      includeTags 'flaky'
     }
 
     develocity {
@@ -945,7 +949,7 @@ project(':server') {
     testImplementation libs.junitJupiter
     testImplementation libs.slf4jReload4j
 
-    testRuntimeOnly libs.junitPlatformLanucher
+    testRuntimeOnly runtimeTestLibs
   }
 
   task createVersionFile() {
@@ -1004,7 +1008,7 @@ project(':share') {
     testImplementation libs.mockitoCore
     testImplementation libs.slf4jReload4j
 
-    testRuntimeOnly libs.junitPlatformLanucher
+    testRuntimeOnly runtimeTestLibs
   }
 
   checkstyle {
@@ -1114,7 +1118,7 @@ project(':core') {
     testImplementation libs.slf4jReload4j
     testImplementation libs.caffeine
 
-    testRuntimeOnly libs.junitPlatformLanucher
+    testRuntimeOnly runtimeTestLibs
   }
 
   if (userEnableTestCoverage) {
@@ -1367,7 +1371,7 @@ project(':metadata') {
     testImplementation project(':raft').sourceSets.test.output
     testImplementation project(':server-common').sourceSets.test.output
 
-    testRuntimeOnly libs.junitPlatformLanucher
+    testRuntimeOnly runtimeTestLibs
 
     generator project(':generator')
   }
@@ -1536,6 +1540,7 @@ project(':group-coordinator') {
 }
 
 project(':test-common') {
+  // Test framework stuff. Implementations that support test-common-api
   base {
     archivesName = "kafka-test-common"
   }
@@ -1564,11 +1569,11 @@ project(':test-common') {
 }
 
 project(':test-common:test-common-api') {
+  // Interfaces, config classes, and other test APIs
   base {
     archivesName = "kafka-test-common-api"
   }
 
-
   dependencies {
     implementation project(':clients')
     implementation project(':core')
@@ -1596,6 +1601,28 @@ project(':test-common:test-common-api') {
   }
 }
 
+project(':test-common:test-common-runtime') {
+  // Runtime-only test code including JUnit extentions
+  base {
+    archivesName = "kafka-test-common-runtime"
+  }
+
+  dependencies {
+    implementation libs.slf4jApi
+    implementation libs.junitPlatformLanucher
+    implementation libs.junitJupiterApi
+    implementation libs.junitJupiter
+  }
+
+  checkstyle {
+    configProperties = checkstyleConfigProperties("import-control-test-common-api.xml")
+  }
+
+  javadoc {
+    enabled = false
+  }
+}
+
 project(':transaction-coordinator') {
   base {
     archivesName = "kafka-transaction-coordinator"
@@ -1789,7 +1816,7 @@ project(':generator') {
     implementation libs.jacksonJaxrsJsonProvider
     testImplementation libs.junitJupiter
 
-    testRuntimeOnly libs.junitPlatformLanucher
+    testRuntimeOnly runtimeTestLibs
   }
 
   javadoc {
@@ -2094,7 +2121,6 @@ project(':server-common') {
     testImplementation libs.mockitoCore
 
     testRuntimeOnly runtimeTestLibs
-    testRuntimeOnly project(":test-common")
   }
 
   task createVersionFile() {
@@ -2325,7 +2351,7 @@ project(':tools:tools-api') {
   dependencies {
     implementation project(':clients')
     testImplementation libs.junitJupiter
-    testRuntimeOnly libs.junitPlatformLanucher
+    testRuntimeOnly runtimeTestLibs
   }
 
   task createVersionFile() {
@@ -2425,9 +2451,8 @@ project(':tools') {
     testImplementation libs.apachedsProtocolLdap
     testImplementation libs.apachedsLdifPartition
 
-    testRuntimeOnly libs.junitPlatformLanucher
+    testRuntimeOnly runtimeTestLibs
     testRuntimeOnly libs.hamcrest
-    testRuntimeOnly project(':test-common')
   }
 
   javadoc {
@@ -2859,7 +2884,7 @@ project(':streams:examples') {
     testImplementation libs.junitJupiter
     testImplementation libs.hamcrest
 
-    testRuntimeOnly libs.junitPlatformLanucher
+    testRuntimeOnly runtimeTestLibs
   }
 
   javadoc {

clients/src/test/java/org/apache/kafka/common/UuidTest.java

@@ -77,7 +77,7 @@ public class UuidTest {
         assertEquals(Uuid.fromString(zeroIdString), Uuid.ZERO_UUID);
     }
 
-    @RepeatedTest(100)
+    @RepeatedTest(value = 100, name = RepeatedTest.LONG_DISPLAY_NAME)
     public void testRandomUuid() {
         Uuid randomID = Uuid.randomUuid();
 

settings.gradle

@@ -118,7 +118,8 @@ include 'clients',
     'transaction-coordinator',
     'trogdor',
     'test-common',
-    'test-common:test-common-api'
+    'test-common:test-common-api',
+    'test-common:test-common-runtime'
 
 project(":storage:api").name = "storage-api"
 rootProject.name = 'kafka'

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/AutoQuarantinedTestFilter.java

@@ -0,0 +1,172 @@
+/*
+ * 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.kafka.common.test.junit;
+
+import org.junit.platform.engine.Filter;
+import org.junit.platform.engine.FilterResult;
+import org.junit.platform.engine.TestDescriptor;
+import org.junit.platform.engine.TestSource;
+import org.junit.platform.engine.support.descriptor.MethodSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+public class AutoQuarantinedTestFilter implements Filter<TestDescriptor> {
+
+    private static final Filter<TestDescriptor> INCLUDE_ALL_TESTS = testDescriptor -> FilterResult.included(null);
+    private static final Filter<TestDescriptor> EXCLUDE_ALL_TESTS = testDescriptor -> FilterResult.excluded(null);
+
+    private static final Logger log = LoggerFactory.getLogger(AutoQuarantinedTestFilter.class);
+
+    private final Set<TestAndMethod> testCatalog;
+    private final boolean includeQuarantined;
+
+    AutoQuarantinedTestFilter(Set<TestAndMethod> testCatalog, boolean includeQuarantined) {
+        this.testCatalog = Collections.unmodifiableSet(testCatalog);
+        this.includeQuarantined = includeQuarantined;
+    }
+
+    @Override
+    public FilterResult apply(TestDescriptor testDescriptor) {
+        Optional<TestSource> sourceOpt = testDescriptor.getSource();
+        if (sourceOpt.isEmpty()) {
+            return FilterResult.included(null);
+        }
+
+        TestSource source = sourceOpt.get();
+        if (!(source instanceof MethodSource)) {
+            return FilterResult.included(null);
+        }
+
+        MethodSource methodSource = (MethodSource) source;
+
+        TestAndMethod testAndMethod = new TestAndMethod(methodSource.getClassName(), methodSource.getMethodName());
+        if (includeQuarantined) {
+            if (testCatalog.contains(testAndMethod)) {
+                return FilterResult.excluded("exclude non-quarantined");
+            } else {
+                return FilterResult.included("auto-quarantined");
+            }
+        } else {
+            if (testCatalog.contains(testAndMethod)) {
+                return FilterResult.included(null);
+            } else {
+                return FilterResult.excluded("auto-quarantined");
+            }
+        }
+    }
+
+    private static Filter<TestDescriptor> defaultFilter(boolean includeQuarantined) {
+        if (includeQuarantined) {
+            return EXCLUDE_ALL_TESTS;
+        } else {
+            return INCLUDE_ALL_TESTS;
+        }
+    }
+
+    /**
+     * Create a filter that excludes tests that are missing from a given test catalog file.
+     * If no test catalog is given, the default behavior depends on {@code includeQuarantined}.
+     * If true, this filter will exclude all tests. If false, this filter will include all tests.
+     * <p>
+     * The format of the test catalog is a text file where each line has the format of:
+     *
+     * <pre>
+     *     FullyQualifiedClassName "#" MethodName "\n"
+     * </pre>
+     *
+     * @param testCatalogFileName path to a test catalog file
+     * @param includeQuarantined true if this filter should include only the auto-quarantined tests
+     */
+    public static Filter<TestDescriptor> create(String testCatalogFileName, boolean includeQuarantined) {
+        if (testCatalogFileName == null || testCatalogFileName.isEmpty()) {
+            log.debug("No test catalog specified, will not quarantine any recently added tests.");
+            return defaultFilter(includeQuarantined);
+        }
+        Path path = Paths.get(testCatalogFileName);
+        log.debug("Loading test catalog file {}.", path);
+
+        if (!Files.exists(path)) {
+            log.error("Test catalog file {} does not exist, will not quarantine any recently added tests.", path);
+            return defaultFilter(includeQuarantined);
+        }
+
+        Set<TestAndMethod> allTests = new HashSet<>();
+        try (BufferedReader reader = Files.newBufferedReader(path, Charset.defaultCharset())) {
+            String line = reader.readLine();
+            while (line != null) {
+                String[] toks = line.split("#", 2);
+                allTests.add(new TestAndMethod(toks[0], toks[1]));
+                line = reader.readLine();
+            }
+        } catch (IOException e) {
+            log.error("Error while reading test catalog file, will not quarantine any recently added tests.", e);
+            return defaultFilter(includeQuarantined);
+        }
+
+        if (allTests.isEmpty()) {
+            log.error("Loaded an empty test catalog, will not quarantine any recently added tests.");
+            return defaultFilter(includeQuarantined);
+        } else {
+            log.debug("Loaded {} test methods from test catalog file {}.", allTests.size(), path);
+            return new AutoQuarantinedTestFilter(allTests, includeQuarantined);
+        }
+    }
+
+    public static class TestAndMethod {
+        private final String testClass;
+        private final String testMethod;
+
+        public TestAndMethod(String testClass, String testMethod) {
+            this.testClass = testClass;
+            this.testMethod = testMethod;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            TestAndMethod that = (TestAndMethod) o;
+            return Objects.equals(testClass, that.testClass) && Objects.equals(testMethod, that.testMethod);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(testClass, testMethod);
+        }
+
+        @Override
+        public String toString() {
+            return "TestAndMethod{" +
+                "testClass='" + testClass + '\'' +
+                ", testMethod='" + testMethod + '\'' +
+                '}';
+        }
+    }
+}

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/QuarantinedPostDiscoveryFilter.java

@@ -0,0 +1,87 @@
+/*
+ * 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.kafka.common.test.junit;
+
+import org.junit.platform.engine.Filter;
+import org.junit.platform.engine.FilterResult;
+import org.junit.platform.engine.TestDescriptor;
+import org.junit.platform.engine.TestTag;
+import org.junit.platform.launcher.PostDiscoveryFilter;
+
+/**
+ * A JUnit test filter which can include or exclude discovered tests before
+ * they are sent off to the test engine for execution. The behavior of this
+ * filter is controlled by the system property "kafka.test.run.quarantined".
+ * If the property is set to "true", then only auto-quarantined and explicitly
+ * {@code @Flaky} tests will be included. If the property is set to "false", then
+ * only non-quarantined tests will be run.
+ * <p>
+ * This filter is registered with JUnit using SPI. The test-common-runtime module
+ * includes a META-INF/services/org.junit.platform.launcher.PostDiscoveryFilter
+ * service file which registers this class.
+ */
+public class QuarantinedPostDiscoveryFilter implements PostDiscoveryFilter {
+
+    private static final TestTag FLAKY_TEST_TAG = TestTag.create("flaky");
+
+    public static final String RUN_QUARANTINED_PROP = "kafka.test.run.quarantined";
+
+    public static final String CATALOG_FILE_PROP = "kafka.test.catalog.file";
+
+    private final Filter<TestDescriptor> autoQuarantinedFilter;
+    private final boolean runQuarantined;
+
+    // No-arg public constructor for SPI
+    @SuppressWarnings("unused")
+    public QuarantinedPostDiscoveryFilter() {
+        runQuarantined = System.getProperty(RUN_QUARANTINED_PROP, "false")
+            .equalsIgnoreCase("true");
+
+        String testCatalogFileName = System.getProperty(CATALOG_FILE_PROP);
+        autoQuarantinedFilter = AutoQuarantinedTestFilter.create(testCatalogFileName, runQuarantined);
+    }
+
+    // Visible for tests
+    QuarantinedPostDiscoveryFilter(Filter<TestDescriptor> autoQuarantinedFilter, boolean runQuarantined) {
+        this.autoQuarantinedFilter = autoQuarantinedFilter;
+        this.runQuarantined = runQuarantined;
+    }
+
+    @Override
+    public FilterResult apply(TestDescriptor testDescriptor) {
+        boolean hasTag = testDescriptor.getTags().contains(FLAKY_TEST_TAG);
+        FilterResult result = autoQuarantinedFilter.apply(testDescriptor);
+        if (runQuarantined) {
+            // If selecting quarantined tests, we first check for explicitly flaky tests. If no
+            // flaky tag is set, check the auto-quarantined filter. In the case of a missing test
+            // catalog, the auto-quarantined filter will exclude all tests.
+            if (hasTag) {
+                return FilterResult.included("flaky");
+            } else {
+                return result;
+            }
+        } else {
+            // If selecting non-quarantined tests, we exclude auto-quarantined tests and flaky tests
+            if (result.included() && hasTag) {
+                return FilterResult.excluded("flaky");
+            } else {
+                return result;
+            }
+        }
+    }
+}

test-common/test-common-runtime/src/main/resources/META-INF/services/org.junit.platform.launcher.PostDiscoveryFilter

@@ -12,4 +12,5 @@
 # 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.
-junit.jupiter.params.displayname.default = "{displayName}.{argumentsWithNames}"
+
+org.apache.kafka.common.test.junit.QuarantinedPostDiscoveryFilter

test-common/test-common-runtime/src/main/resources/junit-platform.properties

@@ -13,3 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 junit.jupiter.params.displayname.default = "{displayName}.{argumentsWithNames}"
+junit.jupiter.extensions.autodetection.enabled = true

test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/junit/AutoQuarantinedTestFilterTest.java

@@ -0,0 +1,82 @@
+/*
+ * 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.kafka.common.test.junit;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.platform.engine.Filter;
+import org.junit.platform.engine.TestDescriptor;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AutoQuarantinedTestFilterTest {
+
+    private TestDescriptor descriptor(String className, String methodName) {
+        return new QuarantinedPostDiscoveryFilterTest.MockTestDescriptor(className, methodName);
+    }
+
+    @Test
+    public void testLoadCatalog(@TempDir Path tempDir) throws IOException {
+        Path catalog = tempDir.resolve("catalog.txt");
+        List<String> lines = new ArrayList<>();
+        lines.add("o.a.k.Foo#testBar1");
+        lines.add("o.a.k.Foo#testBar2");
+        lines.add("o.a.k.Spam#testEggs");
+        Files.write(catalog, lines);
+
+        Filter<TestDescriptor> filter = AutoQuarantinedTestFilter.create(catalog.toString(), false);
+        assertTrue(filter.apply(descriptor("o.a.k.Foo", "testBar1")).included());
+        assertTrue(filter.apply(descriptor("o.a.k.Foo", "testBar2")).included());
+        assertTrue(filter.apply(descriptor("o.a.k.Spam", "testEggs")).included());
+        assertTrue(filter.apply(descriptor("o.a.k.Spam", "testNew")).excluded());
+
+        filter = AutoQuarantinedTestFilter.create(catalog.toString(), true);
+        assertTrue(filter.apply(descriptor("o.a.k.Foo", "testBar1")).excluded());
+        assertTrue(filter.apply(descriptor("o.a.k.Foo", "testBar2")).excluded());
+        assertTrue(filter.apply(descriptor("o.a.k.Spam", "testEggs")).excluded());
+        assertTrue(filter.apply(descriptor("o.a.k.Spam", "testNew")).included());
+    }
+
+    @Test
+    public void testEmptyCatalog(@TempDir Path tempDir) throws IOException {
+        Path catalog = tempDir.resolve("catalog.txt");
+        Files.write(catalog, Collections.emptyList());
+
+        Filter<TestDescriptor> filter = AutoQuarantinedTestFilter.create(catalog.toString(), false);
+        assertTrue(filter.apply(descriptor("o.a.k.Foo", "testBar1")).included());
+        assertTrue(filter.apply(descriptor("o.a.k.Foo", "testBar2")).included());
+        assertTrue(filter.apply(descriptor("o.a.k.Spam", "testEggs")).included());
+        assertTrue(filter.apply(descriptor("o.a.k.Spam", "testNew")).included());
+    }
+
+    @Test
+    public void testMissingCatalog() {
+        Filter<TestDescriptor> filter = AutoQuarantinedTestFilter.create("does-not-exist.txt", false);
+        assertTrue(filter.apply(descriptor("o.a.k.Foo", "testBar1")).included());
+        assertTrue(filter.apply(descriptor("o.a.k.Foo", "testBar2")).included());
+        assertTrue(filter.apply(descriptor("o.a.k.Spam", "testEggs")).included());
+        assertTrue(filter.apply(descriptor("o.a.k.Spam", "testNew")).included());
+    }
+}

test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/junit/QuarantinedPostDiscoveryFilterTest.java

@@ -0,0 +1,175 @@
+/*
+ * 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.kafka.common.test.junit;
+
+import org.junit.jupiter.api.Test;
+import org.junit.platform.engine.TestDescriptor;
+import org.junit.platform.engine.TestSource;
+import org.junit.platform.engine.TestTag;
+import org.junit.platform.engine.UniqueId;
+import org.junit.platform.engine.support.descriptor.MethodSource;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class QuarantinedPostDiscoveryFilterTest {
+
+    static class MockTestDescriptor implements TestDescriptor {
+
+        private final MethodSource methodSource;
+        private final Set<TestTag> testTags;
+
+        MockTestDescriptor(String className, String methodName, String... tags) {
+            this.methodSource = MethodSource.from(className, methodName);
+            this.testTags = new HashSet<>();
+            Arrays.stream(tags).forEach(tag -> testTags.add(TestTag.create(tag)));
+        }
+
+        @Override
+        public UniqueId getUniqueId() {
+            return null;
+        }
+
+        @Override
+        public String getDisplayName() {
+            return "";
+        }
+
+        @Override
+        public Set<TestTag> getTags() {
+            return this.testTags;
+        }
+
+        @Override
+        public Optional<TestSource> getSource() {
+            return Optional.of(this.methodSource);
+        }
+
+        @Override
+        public Optional<TestDescriptor> getParent() {
+            return Optional.empty();
+        }
+
+        @Override
+        public void setParent(TestDescriptor testDescriptor) {
+
+        }
+
+        @Override
+        public Set<? extends TestDescriptor> getChildren() {
+            return Set.of();
+        }
+
+        @Override
+        public void addChild(TestDescriptor testDescriptor) {
+
+        }
+
+        @Override
+        public void removeChild(TestDescriptor testDescriptor) {
+
+        }
+
+        @Override
+        public void removeFromHierarchy() {
+
+        }
+
+        @Override
+        public Type getType() {
+            return null;
+        }
+
+        @Override
+        public Optional<? extends TestDescriptor> findByUniqueId(UniqueId uniqueId) {
+            return Optional.empty();
+        }
+    }
+
+    QuarantinedPostDiscoveryFilter setupFilter(boolean runQuarantined) {
+        Set<AutoQuarantinedTestFilter.TestAndMethod> testCatalog = new HashSet<>();
+        testCatalog.add(new AutoQuarantinedTestFilter.TestAndMethod("o.a.k.Foo", "testBar1"));
+        testCatalog.add(new AutoQuarantinedTestFilter.TestAndMethod("o.a.k.Foo", "testBar2"));
+        testCatalog.add(new AutoQuarantinedTestFilter.TestAndMethod("o.a.k.Spam", "testEggs"));
+
+        AutoQuarantinedTestFilter autoQuarantinedTestFilter = new AutoQuarantinedTestFilter(testCatalog, runQuarantined);
+        return new QuarantinedPostDiscoveryFilter(autoQuarantinedTestFilter, runQuarantined);
+    }
+
+    @Test
+    public void testQuarantinedExistingTestNonFlaky() {
+        QuarantinedPostDiscoveryFilter filter = setupFilter(true);
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar1")).excluded());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar2")).excluded());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Spam", "testEggs")).excluded());
+    }
+
+    @Test
+    public void testQuarantinedExistingTestFlaky() {
+        QuarantinedPostDiscoveryFilter filter = setupFilter(true);
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar1", "flaky")).included());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar2", "flaky")).included());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Spam", "testEggs", "flaky", "integration")).included());
+    }
+
+    @Test
+    public void testQuarantinedNewTest() {
+        QuarantinedPostDiscoveryFilter filter = setupFilter(true);
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar3")).included());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Spam", "testEggz", "flaky")).included());
+    }
+
+    @Test
+    public void testExistingTestNonFlaky() {
+        QuarantinedPostDiscoveryFilter filter = setupFilter(false);
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar1")).included());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar2")).included());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Spam", "testEggs")).included());
+    }
+
+
+    @Test
+    public void testExistingTestFlaky() {
+        QuarantinedPostDiscoveryFilter filter = setupFilter(false);
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar1", "flaky")).excluded());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar2", "flaky")).excluded());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Spam", "testEggs", "flaky", "integration")).excluded());
+    }
+
+    @Test
+    public void testNewTest() {
+        QuarantinedPostDiscoveryFilter filter = setupFilter(false);
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar3")).excluded());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Spam", "testEggz", "flaky")).excluded());
+    }
+
+    @Test
+    public void testNoCatalogQuarantinedTest() {
+        QuarantinedPostDiscoveryFilter filter = new QuarantinedPostDiscoveryFilter(
+            AutoQuarantinedTestFilter.create(null, true),
+            true
+        );
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar1", "flaky")).included());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Foo", "testBar2", "flaky")).included());
+        assertTrue(filter.apply(new MockTestDescriptor("o.a.k.Spam", "testEggs")).excluded());
+    }
+}