This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12812-c0d2398c86098062e7d6e40a847d97423301ea3c in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit cce8e199e5adc976742c94e5ba32dad73aafc3b7 Author: zhiweiniu <[email protected]> AuthorDate: Fri Oct 16 08:48:08 2026 +0000 [Improve][CI] Run only the E2E engines affected by the change (#12812) --- .github/workflows/backend.yml | 146 ++++++++++++++++++++- .../seatunnel/e2e/common/util/ContainerUtil.java | 111 +++++++++++++--- .../e2e/common/util/ContainerUtilTest.java | 105 +++++++++++++++ tools/update_modules_check/run_engines.py | 99 ++++++++++++++ tools/update_modules_check/test_run_engines.py | 125 ++++++++++++++++++ 5 files changed, 559 insertions(+), 27 deletions(-) diff --git a/.github/workflows/backend.yml b/.github/workflows/backend.yml index ad900a484f..6a26ce2f29 100644 --- a/.github/workflows/backend.yml +++ b/.github/workflows/backend.yml @@ -207,6 +207,12 @@ jobs: timeout-minutes: 20 outputs: api: ${{ steps.filter.outputs.api }} + shared: ${{ steps.filter.outputs.shared }} + translation: ${{ steps.filter.outputs.translation }} + translation-base: ${{ steps.filter.outputs.translation-base }} + translation-flink: ${{ steps.filter.outputs.translation-flink }} + translation-spark: ${{ steps.filter.outputs.translation-spark }} + run-engines: ${{ steps.run-engines.outputs.engines || '["flink","spark","zeta"]' }} engine: ${{ steps.filter.outputs.engine }} engine-e2e: ${{ steps.filter.outputs.engine-e2e }} edge-agent: ${{ steps.filter.outputs.edge-agent }} @@ -319,23 +325,86 @@ jobs: echo "standalone=$true_or_false" >> $GITHUB_OUTPUT echo "standalone_files=$file_list" >> $GITHUB_OUTPUT - api_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "seatunnel-api/**" "seatunnel-common/**" "seatunnel-config/**" "seatunnel-core/**" "seatunnel-e2e/seatunnel-e2e-common/**" "seatunnel-formats/**" "seatunnel-plugin-discovery/**" "seatunnel-transforms-v2/**" "seatunnel-translation/**" "seatunnel-e2e/seatunnel-transforms-v2-e2e/**" "pom.xml" "**/workflows/**" "tools/**" "seatunnel-dist/**"` - true_or_false=${api_files%%$'\n'*} - file_list=${api_files#*$'\n'} + shared_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "seatunnel-api/**" "seatunnel-common/**" "seatunnel-config/**" "seatunnel-core/**" "seatunnel-e2e/seatunnel-e2e-common/**" "seatunnel-formats/**" "seatunnel-plugin-discovery/**" "seatunnel-transforms-v2/**" "seatunnel-e2e/seatunnel-transforms-v2-e2e/**" "pom.xml" "**/workflows/**" "tools/**" "seatunnel-dist/**"` + shared=${shared_files%%$'\n'*} + shared_file_list=${shared_files#*$'\n'} + + # Broad probe: any translation change keeps the API trigger. + translation_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "seatunnel-translation/**"` + translation=${translation_files%%$'\n'*} + + # The broad glob also matches the engine sub-modules, so the engine scope uses narrowed + # probes: the base module is shared by the Flink and Spark translators, while the Flink + # and Spark translators are engine specific. + translation_base_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "seatunnel-translation/pom.xml" "seatunnel-translation/seatunnel-translation-base/**"` + translation_base=${translation_base_files%%$'\n'*} + + translation_flink_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "seatunnel-translation/seatunnel-translation-flink/**"` + translation_flink=${translation_flink_files%%$'\n'*} + + translation_spark_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "seatunnel-translation/seatunnel-translation-spark/**"` + translation_spark=${translation_spark_files%%$'\n'*} + # Apache dev, main, master, and numeric *-release pushes force full API coverage. # On fork pushes, workflow files, tools/update_modules_check, seatunnel-dist, and # bin/install-plugin.sh do not trigger the full connector matrix by themselves. ci_scope_ref="${GITHUB_BASE_REF:-$GITHUB_REF}" - if ! true_or_false=$(python tools/update_modules_check/ci_scope.py "$repository_owner" "$ci_scope_ref" --api-changed "$true_or_false" --api-files-json "$file_list"); then + if ! shared=$(python tools/update_modules_check/ci_scope.py "$repository_owner" "$ci_scope_ref" --api-changed "$shared" --api-files-json "$shared_file_list"); then echo "::warning::CI scope helper failed; forcing full API coverage" - true_or_false='true' + shared='true' fi - if [[ $true_or_false != 'true' && $true_or_false != 'false' ]]; then + if [[ $shared != 'true' && $shared != 'false' ]]; then echo "::warning::CI scope helper returned an invalid value; forcing full API coverage" + shared='true' + fi + # The translation adapters are the only API-scoped code that is not shared by every + # engine, so the API trigger is the union of the shared and the translation groups. + true_or_false='false' + if [[ $shared == 'true' || $translation == 'true' ]]; then true_or_false='true' fi + echo "shared=$shared" >> $GITHUB_OUTPUT + echo "shared_files=$shared_file_list" >> $GITHUB_OUTPUT + echo "translation=$translation" >> $GITHUB_OUTPUT + echo "translation-base=$translation_base" >> $GITHUB_OUTPUT + echo "translation-flink=$translation_flink" >> $GITHUB_OUTPUT + echo "translation-spark=$translation_spark" >> $GITHUB_OUTPUT echo "api=$true_or_false" >> $GITHUB_OUTPUT - echo "api_files=$file_list" >> $GITHUB_OUTPUT + echo "api_files=$shared_file_list" >> $GITHUB_OUTPUT + + - name: Decide engines to run + id: run-engines + env: + SHARED: ${{ steps.filter.outputs.shared }} + CONNECTORS: ${{ steps.filter.outputs.cv2 }} + CONNECTOR_E2E: ${{ steps.filter.outputs.cv2-e2e }} + DIST: ${{ steps.filter.outputs.dist }} + STANDALONE: ${{ steps.filter.outputs.standalone }} + TRANSLATION_BASE: ${{ steps.filter.outputs.translation-base }} + TRANSLATION_FLINK: ${{ steps.filter.outputs.translation-flink }} + TRANSLATION_SPARK: ${{ steps.filter.outputs.translation-spark }} + ENGINE: ${{ steps.filter.outputs.engine }} + ENGINE_E2E: ${{ steps.filter.outputs.engine-e2e }} + run: | + # Every E2E job passes this list to the tests as RUN_ENGINES, so a job only starts the + # containers of the engines the change can affect. Unknown or empty scopes keep every + # engine, and tools/update_modules_check/run_engines.py owns the mapping rules. + if ! engines=$(python tools/update_modules_check/run_engines.py \ + --shared "${SHARED:-false}" \ + --connectors "${CONNECTORS:-false}" \ + --connector-e2e "${CONNECTOR_E2E:-false}" \ + --dist "${DIST:-false}" \ + --standalone "${STANDALONE:-false}" \ + --translation-base "${TRANSLATION_BASE:-false}" \ + --translation-flink "${TRANSLATION_FLINK:-false}" \ + --translation-spark "${TRANSLATION_SPARK:-false}" \ + --engine "${ENGINE:-false}" \ + --engine-e2e "${ENGINE_E2E:-false}"); then + echo "::warning::Engine scope helper failed; running every engine" + engines='["flink","spark","zeta"]' + fi + echo "Run engines: $engines" + echo "engines=$engines" >> "$GITHUB_OUTPUT" - name: Check Connector V2 Update id: cv2-modules @@ -695,6 +764,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'false' && needs.changes.outputs.engine == 'false' && needs.changes.outputs.it-modules != '[]' && contains(fromJSON(needs.changes.outputs.updated-it-shards), 'part-1') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -733,6 +804,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'false' && needs.changes.outputs.engine == 'false' && needs.changes.outputs.it-modules != '[]' && contains(fromJSON(needs.changes.outputs.updated-it-shards), 'part-2') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -771,6 +844,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'false' && needs.changes.outputs.engine == 'false' && needs.changes.outputs.it-modules != '[]' && contains(fromJSON(needs.changes.outputs.updated-it-shards), 'part-3') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -809,6 +884,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'false' && needs.changes.outputs.engine == 'false' && needs.changes.outputs.it-modules != '[]' && contains(fromJSON(needs.changes.outputs.updated-it-shards), 'part-4') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -846,6 +923,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'false' && needs.changes.outputs.engine == 'false' && needs.changes.outputs.it-modules != '[]' && contains(fromJSON(needs.changes.outputs.updated-it-shards), 'part-5') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -883,6 +962,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'false' && needs.changes.outputs.engine == 'false' && needs.changes.outputs.it-modules != '[]' && contains(fromJSON(needs.changes.outputs.updated-it-shards), 'part-6') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -920,6 +1001,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'false' && needs.changes.outputs.engine == 'false' && needs.changes.outputs.it-modules != '[]' && contains(fromJSON(needs.changes.outputs.updated-it-shards), 'part-7') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -958,6 +1041,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'false' && needs.changes.outputs.engine == 'false' && needs.changes.outputs.it-modules != '[]' && contains(fromJSON(needs.changes.outputs.updated-it-shards), 'part-8') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1101,6 +1186,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1136,6 +1222,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1174,6 +1261,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1211,6 +1299,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1248,6 +1337,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1285,6 +1375,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1322,6 +1413,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1359,6 +1451,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1396,6 +1489,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1433,6 +1527,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1467,6 +1562,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1501,6 +1597,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1538,6 +1635,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1572,6 +1670,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1606,6 +1705,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1640,6 +1740,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1674,6 +1775,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1708,6 +1810,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1742,6 +1845,7 @@ jobs: env: RUN_ALL_CONTAINER: ${{ needs.changes.outputs.api }} RUN_ZETA_CONTAINER: ${{ needs.changes.outputs.engine }} + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1773,6 +1877,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-kudu-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1805,6 +1911,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-amazonsqs-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1836,6 +1944,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-google-pubsub-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1867,6 +1977,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-kafka-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1898,6 +2010,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-rocketmq-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1929,6 +2043,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-elasticsearch-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1960,6 +2076,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-cdc-mysql-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -1991,6 +2109,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-doris-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -2022,6 +2142,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-paimon-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -2053,6 +2175,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-cdc-oracle-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -2089,6 +2213,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-file-local-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -2120,6 +2246,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-file-sftp-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -2151,6 +2279,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-redis-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: @@ -2182,6 +2312,8 @@ jobs: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'true' || needs.changes.outputs.engine == 'true' || contains(fromJSON(needs.changes.outputs.it-modules), 'connector-sensorsdata-e2e') runs-on: ${{ matrix.os }} + env: + RUN_ENGINES: ${{ needs.changes.outputs.run-engines }} strategy: fail-fast: false matrix: diff --git a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/util/ContainerUtil.java b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/util/ContainerUtil.java index 4b27edf21d..7502d23a3c 100644 --- a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/util/ContainerUtil.java +++ b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/util/ContainerUtil.java @@ -51,6 +51,7 @@ import java.util.Arrays; import java.util.HashSet; import java.util.LinkedList; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.ServiceConfigurationError; @@ -69,6 +70,14 @@ public final class ContainerUtil { public static final String PLUGIN_MAPPING_FILE = "plugin-mapping.properties"; + /** + * Engines whose containers an E2E suite runs, for example {@code RUN_ENGINES=zeta} or {@code + * RUN_ENGINES=["flink","spark"]}. CI sets it from the changed modules and developers use it to + * verify a suite locally against a single engine. Leaving it unset, blank, or {@code all} keeps + * every engine, while an unknown engine name fails fast with a clear error. + */ + public static final String RUN_ENGINES = "RUN_ENGINES"; + /** An error occurs when the user is not a submodule of seatunnel-e2e. */ public static final String PROJECT_ROOT_PATH = getProjectRootPath(); @@ -401,27 +410,28 @@ public final class ContainerUtil { testAllContainer, testZetaContainer); if (isTestInPR) { - return result.stream() - .filter(container -> container.identifier().isTestInPR()) - .filter( - container -> { - if (testAllContainer - || container.identifier().equals(FLINK_1_18) - || container.identifier().equals(FLINK_1_20) - || container.identifier().equals(SPARK_3_3)) { - return true; - } - if (testZetaContainer) { - return container - .identifier() - .getEngineType() - .equals(EngineType.SEATUNNEL); - } - return true; - }) - .collect(Collectors.toList()); + return filterByRunEngines( + result.stream() + .filter(container -> container.identifier().isTestInPR()) + .filter( + container -> { + if (testAllContainer + || container.identifier().equals(FLINK_1_18) + || container.identifier().equals(FLINK_1_20) + || container.identifier().equals(SPARK_3_3)) { + return true; + } + if (testZetaContainer) { + return container + .identifier() + .getEngineType() + .equals(EngineType.SEATUNNEL); + } + return true; + }) + .collect(Collectors.toList())); } else { - return result; + return filterByRunEngines(result); } } catch (ServiceConfigurationError e) { log.error("Could not load service provider for containers.", e); @@ -429,6 +439,67 @@ public final class ContainerUtil { } } + /** + * Keeps the containers of the engines listed in {@code RUN_ENGINES}. + * + * <p>When the variable is unset, blank, {@code all}, or matches none of the discovered + * containers, every discovered container is kept, so an engine cannot silently disappear from a + * suite. An engine name that cannot be parsed fails fast instead of being ignored. + */ + private static List<TestContainer> filterByRunEngines(List<TestContainer> containers) { + return filterByRunEngines(containers, System.getenv(RUN_ENGINES)); + } + + static List<TestContainer> filterByRunEngines( + List<TestContainer> containers, String rawRunEngines) { + if (StringUtils.isBlank(rawRunEngines)) { + return containers; + } + Set<EngineType> engines = parseRunEngines(rawRunEngines); + if (engines.isEmpty()) { + return containers; + } + List<TestContainer> selected = + containers.stream() + .filter( + container -> + engines.contains(container.identifier().getEngineType())) + .collect(Collectors.toList()); + if (selected.isEmpty()) { + log.warn( + "RUN_ENGINES[{}] matches none of the discovered containers, run them all", + rawRunEngines); + return containers; + } + log.info("Run the containers of the engines: {}", rawRunEngines); + return selected; + } + + /** Parses {@code RUN_ENGINES}: engine names in any case, separated by commas. */ + private static Set<EngineType> parseRunEngines(String rawRunEngines) { + Set<EngineType> engines = new HashSet<>(); + for (String engine : rawRunEngines.replace("[", "").replace("]", "").split(",")) { + String name = engine.replace("\"", "").replace("'", "").trim(); + if (name.isEmpty() || "all".equalsIgnoreCase(name)) { + continue; + } + if ("zeta".equalsIgnoreCase(name) || "seatunnel".equalsIgnoreCase(name)) { + engines.add(EngineType.SEATUNNEL); + continue; + } + try { + engines.add(EngineType.valueOf(name.toUpperCase(Locale.ROOT))); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException( + "Unsupported engine '" + + name + + "' in RUN_ENGINES, expected flink, spark, zeta or all", + e); + } + } + return engines; + } + public static void copyFileIntoContainers( String fileName, String targetPath, GenericContainer<?> container) { Path path = getResourcesFile(fileName).toPath(); diff --git a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/util/ContainerUtilTest.java b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/util/ContainerUtilTest.java new file mode 100644 index 0000000000..b324143d19 --- /dev/null +++ b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/util/ContainerUtilTest.java @@ -0,0 +1,105 @@ +/* + * 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.seatunnel.e2e.common.util; + +import org.apache.seatunnel.e2e.common.container.TestContainer; +import org.apache.seatunnel.e2e.common.container.TestContainerId; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class ContainerUtilTest { + + private static final List<TestContainer> CONTAINERS = + Arrays.asList( + container(TestContainerId.FLINK_1_18), + container(TestContainerId.FLINK_1_20), + container(TestContainerId.SPARK_3_3), + container(TestContainerId.SEATUNNEL)); + + @Test + void shouldKeepEveryContainerWhenRunEnginesIsUnset() { + assertEquals(CONTAINERS, ContainerUtil.filterByRunEngines(CONTAINERS, null)); + assertEquals(CONTAINERS, ContainerUtil.filterByRunEngines(CONTAINERS, " ")); + assertEquals(CONTAINERS, ContainerUtil.filterByRunEngines(CONTAINERS, "[]")); + assertEquals(CONTAINERS, ContainerUtil.filterByRunEngines(CONTAINERS, "all")); + } + + @Test + void shouldKeepOnlyTheContainersOfTheListedEngines() { + assertEquals( + Arrays.asList(TestContainerId.FLINK_1_18, TestContainerId.FLINK_1_20), + identifiers(ContainerUtil.filterByRunEngines(CONTAINERS, "flink"))); + assertEquals( + Arrays.asList(TestContainerId.SPARK_3_3), + identifiers(ContainerUtil.filterByRunEngines(CONTAINERS, "'spark'"))); + assertEquals( + Arrays.asList( + TestContainerId.FLINK_1_18, + TestContainerId.FLINK_1_20, + TestContainerId.SPARK_3_3), + identifiers( + ContainerUtil.filterByRunEngines(CONTAINERS, "[\"flink\", \"spark\"]"))); + } + + @Test + void shouldAcceptTheZetaAliases() { + assertEquals( + Arrays.asList(TestContainerId.SEATUNNEL), + identifiers(ContainerUtil.filterByRunEngines(CONTAINERS, "zeta"))); + assertEquals( + Arrays.asList(TestContainerId.SEATUNNEL), + identifiers(ContainerUtil.filterByRunEngines(CONTAINERS, "SEATUNNEL"))); + } + + @Test + void shouldKeepEveryContainerWhenNoDiscoveredContainerMatches() { + List<TestContainer> zetaOnly = Arrays.asList(container(TestContainerId.SEATUNNEL)); + + assertEquals(zetaOnly, ContainerUtil.filterByRunEngines(zetaOnly, "flink,spark")); + } + + @Test + void shouldFailFastForUnknownEngineNames() { + IllegalArgumentException failure = + assertThrows( + IllegalArgumentException.class, + () -> ContainerUtil.filterByRunEngines(CONTAINERS, "zetta")); + + assertTrue(failure.getMessage().contains("zetta"), failure.getMessage()); + } + + private static List<TestContainerId> identifiers(List<TestContainer> containers) { + return containers.stream().map(TestContainer::identifier).collect(Collectors.toList()); + } + + private static TestContainer container(TestContainerId identifier) { + TestContainer container = mock(TestContainer.class); + when(container.identifier()).thenReturn(identifier); + return container; + } +} diff --git a/tools/update_modules_check/run_engines.py b/tools/update_modules_check/run_engines.py new file mode 100644 index 0000000000..93bb82ba7d --- /dev/null +++ b/tools/update_modules_check/run_engines.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +# 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. + +"""Decide which engines the backend integration test jobs have to cover. + +The backend workflow runs the integration test suites against the Flink, Spark and Zeta +containers. Every job passes the resulting list to the tests through the ``RUN_ENGINES`` +environment variable, so a job only starts the containers of the engines a change can +affect: + +* shared code, connector modules, the distribution and the standalone modules are used by + every engine, so they keep the full matrix; +* ``seatunnel-translation`` implements the plan translators of the Flink and Spark + starters only, so it never needs the Zeta container: the base module is shared by both, + while ``seatunnel-translation-flink`` / ``-spark`` only need their own engine; +* ``seatunnel-engine`` is the Zeta engine and is not used by the Flink and Spark starters; +* when no rule matches, every engine runs: coverage is only narrowed on purpose. +""" + +import argparse +import json +from typing import Iterable, List + + +ALL_ENGINES = ("flink", "spark", "zeta") + +# Engines a path group can affect. Keep this in sync with the globs of the `changes` job in +# .github/workflows/backend.yml. +GROUP_ENGINES = { + "shared": ALL_ENGINES, + "connectors": ALL_ENGINES, + "connector-e2e": ALL_ENGINES, + "dist": ALL_ENGINES, + "standalone": ALL_ENGINES, + "translation-base": ("flink", "spark"), + "translation-flink": ("flink",), + "translation-spark": ("spark",), + "engine": ("zeta",), + "engine-e2e": ("zeta",), +} + + +def run_engines_for_groups(changed_groups: Iterable[str]) -> List[str]: + """Return the engines to run for the changed path groups.""" + + engines = set() + for group in changed_groups: + engines.update(GROUP_ENGINES[group]) + if not engines: + return list(ALL_ENGINES) + return [engine for engine in ALL_ENGINES if engine in engines] + + +def parse_bool(value: str) -> bool: + """Parse the shell-friendly booleans emitted by the workflow.""" + + if value == "true": + return True + if value == "false": + return False + raise ValueError(f"invalid boolean: {value}") + + +def main() -> None: + """Print the JSON list of engines for the workflow caller.""" + + parser = argparse.ArgumentParser( + description="Decide which engines the backend integration test jobs have to cover." + ) + for group in GROUP_ENGINES: + parser.add_argument( + f"--{group}", + choices=("true", "false"), + default="false", + help=f"Whether the `{group}` path group changed", + ) + args = parser.parse_args() + + changed_groups = [ + group for group in GROUP_ENGINES if parse_bool(getattr(args, group.replace("-", "_"))) + ] + print(json.dumps(run_engines_for_groups(changed_groups), separators=(",", ":"))) + + +if __name__ == "__main__": + main() diff --git a/tools/update_modules_check/test_run_engines.py b/tools/update_modules_check/test_run_engines.py new file mode 100644 index 0000000000..8ee9806d41 --- /dev/null +++ b/tools/update_modules_check/test_run_engines.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +# 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. + + +import re +import subprocess +import sys +import unittest +from pathlib import Path + +from run_engines import GROUP_ENGINES, run_engines_for_groups + + +WORKFLOW_PATH = ( + Path(__file__).resolve().parents[2] / ".github" / "workflows" / "backend.yml" +) + + +class RunEnginesTest(unittest.TestCase): + """Verify that the E2E engine scope follows the changed path groups.""" + + def test_translation_base_change_runs_flink_and_spark(self) -> None: + self.assertEqual(["flink", "spark"], run_engines_for_groups(["translation-base"])) + + def test_translation_flink_change_runs_flink_only(self) -> None: + self.assertEqual(["flink"], run_engines_for_groups(["translation-flink"])) + + def test_translation_spark_change_runs_spark_only(self) -> None: + self.assertEqual(["spark"], run_engines_for_groups(["translation-spark"])) + + def test_translation_base_and_flink_change_run_both_engines(self) -> None: + self.assertEqual( + ["flink", "spark"], + run_engines_for_groups(["translation-base", "translation-flink"]), + ) + + def test_engine_change_runs_zeta_only(self) -> None: + self.assertEqual(["zeta"], run_engines_for_groups(["engine"])) + + def test_engine_e2e_change_runs_zeta_only(self) -> None: + self.assertEqual(["zeta"], run_engines_for_groups(["engine-e2e"])) + + def test_engine_and_translation_change_run_every_engine(self) -> None: + self.assertEqual( + ["flink", "spark", "zeta"], + run_engines_for_groups(["engine", "translation-base"]), + ) + + def test_engine_and_connector_change_run_every_engine(self) -> None: + self.assertEqual( + ["flink", "spark", "zeta"], + run_engines_for_groups(["engine", "connectors"]), + ) + + def test_shared_changes_run_every_engine(self) -> None: + for group in ("shared", "connectors", "connector-e2e", "dist", "standalone"): + with self.subTest(group=group): + self.assertEqual( + ["flink", "spark", "zeta"], run_engines_for_groups([group]) + ) + + def test_unknown_change_runs_every_engine(self) -> None: + self.assertEqual(["flink", "spark", "zeta"], run_engines_for_groups([])) + + +class RunEnginesWorkflowTest(unittest.TestCase): + """Verify the workflow wiring and the CLI stay in step with the mapping table.""" + + def run_helper(self, *args: str) -> str: + result = subprocess.run( + [sys.executable, str(Path(__file__).resolve().parent / "run_engines.py"), *args], + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + def test_flink_only_translation_change_prints_flink(self) -> None: + self.assertEqual( + '["flink"]', + self.run_helper( + "--translation-base", + "false", + "--translation-flink", + "true", + "--translation-spark", + "false", + ), + ) + + def test_no_group_prints_every_engine(self) -> None: + self.assertEqual('["flink","spark","zeta"]', self.run_helper()) + + def test_workflow_passes_every_group_flag(self) -> None: + invocation = re.search( + r"run_engines\.py(.*?)\); then", WORKFLOW_PATH.read_text(), re.DOTALL + ).group(1) + passed_groups = set(re.findall(r"--([a-z0-9-]+)\s+\"\$\{", invocation)) + self.assertEqual(set(GROUP_ENGINES), passed_groups) + + def test_translation_probes_stay_narrow(self) -> None: + # The broad `seatunnel-translation/**` glob also matches the Flink and Spark + # sub-modules, which would make the flink-only/spark-only scoping unreachable. + base_probe = re.search( + r"translation_base_files=`([^`]*)`", WORKFLOW_PATH.read_text() + ).group(1) + self.assertIn("seatunnel-translation-base/**", base_probe) + self.assertNotIn('"seatunnel-translation/**"', base_probe) + + +if __name__ == "__main__": + unittest.main()
