This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new cce8e199e5 [Improve][CI] Run only the E2E engines affected by the
change (#12812)
cce8e199e5 is described below
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()
