diff --git a/.github/actions/setup-default-test-properties/test-properties.json b/.github/actions/setup-default-test-properties/test-properties.json index f06de5174e6c..7a3f9890a294 100644 --- a/.github/actions/setup-default-test-properties/test-properties.json +++ b/.github/actions/setup-default-test-properties/test-properties.json @@ -14,7 +14,7 @@ }, "JavaTestProperties": { "SUPPORTED_VERSIONS": ["8", "11", "17", "21", "25"], - "FLINK_VERSIONS": ["1.17", "1.18", "1.19", "1.20", "2.0"], + "FLINK_VERSIONS": ["1.17", "1.18", "1.19", "1.20", "2.0", "2.1"], "SPARK_VERSIONS": ["3"] }, "GoTestProperties": { diff --git a/.github/trigger_files/beam_PostCommit_Go.json b/.github/trigger_files/beam_PostCommit_Go.json index 7ab7bcd9a9c6..37dd25bf9029 100644 --- a/.github/trigger_files/beam_PostCommit_Go.json +++ b/.github/trigger_files/beam_PostCommit_Go.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 2 + "modification": 3 } diff --git a/.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Dataflow.json b/.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Dataflow.json index e0266d62f2e0..f1ba03a243ee 100644 --- a/.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Dataflow.json +++ b/.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Dataflow.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 4 + "modification": 5 } diff --git a/.github/workflows/run_rc_validation_java_quickstart.yml b/.github/workflows/run_rc_validation_java_quickstart.yml index dce9b7f3fedb..41a6991d14ee 100644 --- a/.github/workflows/run_rc_validation_java_quickstart.yml +++ b/.github/workflows/run_rc_validation_java_quickstart.yml @@ -88,7 +88,7 @@ jobs: - name: Run QuickStart Java Flink Runner uses: ./.github/actions/gradle-command-self-hosted-action with: - gradle-command: :runners:flink:2.0:runQuickstartJavaFlinkLocal + gradle-command: :runners:flink:2.1:runQuickstartJavaFlinkLocal arguments: | -Prepourl=${{ env.APACHE_REPO_URL }} \ -Pver=${{ env.RELEASE_VERSION }} diff --git a/.test-infra/validate-runner/build.gradle b/.test-infra/validate-runner/build.gradle index 1817d7014a6c..3992abb24dd4 100644 --- a/.test-infra/validate-runner/build.gradle +++ b/.test-infra/validate-runner/build.gradle @@ -31,6 +31,19 @@ repositories { } } +// Flink 2.1+ uses at.yawk.lz4:lz4-java while Spark uses org.lz4:lz4-java +// Resolve capability conflict by preferring Flink's version when both are present +configurations.all { + resolutionStrategy.capabilitiesResolution.withCapability('org.lz4:lz4-java') { + def candidate = candidates.find { it.id.toString().contains('at.yawk.lz4') } + if (candidate != null) { + select(candidate) + } else { + selectHighestVersion() + } + } +} + dependencies { implementation 'com.offbytwo.jenkins:jenkins-client:0.3.8' implementation library.java.jackson_databind diff --git a/examples/java/common.gradle b/examples/java/common.gradle index 10ea43628bc8..0e800e7cf987 100644 --- a/examples/java/common.gradle +++ b/examples/java/common.gradle @@ -35,6 +35,16 @@ configurations.sparkRunnerPreCommit { exclude group: "org.slf4j", module: "jul-to-slf4j" exclude group: "org.slf4j", module: "slf4j-jdk14" } +configurations.flinkRunnerPreCommit { + resolutionStrategy.capabilitiesResolution.withCapability("org.lz4:lz4-java") { + def candidate = candidates.find { it.id.toString().contains('at.yawk.lz4') } + if (candidate != null) { + select(candidate) + } else { + selectHighestVersion() + } + } +} dependencies { directRunnerPreCommit project(path: ":runners:direct-java", configuration: "shadow") diff --git a/gradle.properties b/gradle.properties index 95e50105a494..0289d02722b3 100644 --- a/gradle.properties +++ b/gradle.properties @@ -39,7 +39,7 @@ docker_image_default_repo_root=apache docker_image_default_repo_prefix=beam_ # supported flink versions -flink_versions=1.17,1.18,1.19,1.20,2.0 +flink_versions=1.17,1.18,1.19,1.20,2.0,2.1 # supported spark versions spark_versions=3,4 # supported python versions diff --git a/runners/flink/2.1/build.gradle b/runners/flink/2.1/build.gradle new file mode 100644 index 000000000000..e9092c2977f7 --- /dev/null +++ b/runners/flink/2.1/build.gradle @@ -0,0 +1,56 @@ +/* + * 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. + */ + +project.ext { + flink_major = '2.1' + flink_version = '2.1.3' + excluded_files = [ + 'main': [ + // Used by DataSet API only + "org/apache/beam/runners/flink/adapter/BeamFlinkDataSetAdapter.java", + "org/apache/beam/runners/flink/FlinkBatchPipelineTranslator.java", + "org/apache/beam/runners/flink/FlinkBatchPortablePipelineTranslator.java", + "org/apache/beam/runners/flink/FlinkBatchTransformTranslators.java", + "org/apache/beam/runners/flink/translation/functions/FlinkNonMergingReduceFunction.java", + // Moved to org.apache.flink.runtime.state.StateBackendFactory + "org/apache/beam/runners/flink/FlinkStateBackendFactory.java", + ], + 'test': [ + // Used by DataSet API only + "org/apache/beam/runners/flink/adapter/BeamFlinkDataSetAdapterTest.java", + "org/apache/beam/runners/flink/batch/NonMergingGroupByKeyTest.java", + "org/apache/beam/runners/flink/batch/ReshuffleTest.java", + ] + ] +} + +// Load the main build script which contains all build logic. +apply from: "../flink_runner.gradle" + +// Flink 2.1 uses at.yawk.lz4:lz4-java instead of org.lz4:lz4-java +// Explicitly prefer Flink's at.yawk.lz4 version to resolve capability conflict +configurations.all { + resolutionStrategy.capabilitiesResolution.withCapability('org.lz4:lz4-java') { + def candidate = candidates.find { it.id.toString().contains('at.yawk.lz4') } + if (candidate != null) { + select(candidate) + } else { + selectHighestVersion() + } + } +} diff --git a/runners/flink/2.1/job-server-container/build.gradle b/runners/flink/2.1/job-server-container/build.gradle new file mode 100644 index 000000000000..afdb68a0fc91 --- /dev/null +++ b/runners/flink/2.1/job-server-container/build.gradle @@ -0,0 +1,26 @@ +/* + * 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. + */ + +def basePath = '../../job-server-container' + +project.ext { + resource_path = basePath +} + +// Load the main build script which contains all build logic. +apply from: "$basePath/flink_job_server_container.gradle" diff --git a/runners/flink/2.1/job-server/build.gradle b/runners/flink/2.1/job-server/build.gradle new file mode 100644 index 000000000000..277ddc07fdaf --- /dev/null +++ b/runners/flink/2.1/job-server/build.gradle @@ -0,0 +1,44 @@ +/* + * 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. + */ + +def basePath = '../../job-server' + +project.ext { + // Look for the source code in the parent module + main_source_dirs = ["$basePath/src/main/java"] + test_source_dirs = ["$basePath/src/test/java"] + main_resources_dirs = ["$basePath/src/main/resources"] + test_resources_dirs = ["$basePath/src/test/resources"] + archives_base_name = 'beam-runners-flink-2.1-job-server' +} + +// Load the main build script which contains all build logic. +apply from: "$basePath/flink_job_server.gradle" + +// Flink 2.1 uses at.yawk.lz4:lz4-java instead of org.lz4:lz4-java +// Explicitly prefer Flink's at.yawk.lz4 version to resolve capability conflict +configurations.all { + resolutionStrategy.capabilitiesResolution.withCapability('org.lz4:lz4-java') { + def candidate = candidates.find { it.id.toString().contains('at.yawk.lz4') } + if (candidate != null) { + select(candidate) + } else { + selectHighestVersion() + } + } +} diff --git a/scripts/beam-sql.sh b/scripts/beam-sql.sh index c9ec365b99bc..9df48f6c18d4 100755 --- a/scripts/beam-sql.sh +++ b/scripts/beam-sql.sh @@ -36,6 +36,8 @@ MAVEN_DISTRIBUTION_URL="https://repo.maven.apache.org/maven2/org/apache/maven/ap # Maven Plugin Configuration MAVEN_SHADE_PLUGIN_VERSION="3.5.1" +ICEBERG_VERSION="${ICEBERG_VERSION:-1.10.0}" +APACHE_SNAPSHOT_REPOSITORY_URL="https://repository.apache.org/content/repositories/snapshots/" mkdir -p "${CACHE_DIR}" # Create a temporary directory for our Maven project. @@ -84,6 +86,38 @@ function setup_maven_wrapper() { MAVEN_CMD="${mvnw_script}" } +function add_beam_dependency() { + local artifact_id="$1" + cat >> "${POM_FILE}" << EOL + + org.apache.beam + ${artifact_id} + \${beam.version} + +EOL +} + +function add_dependency() { + local group_id="$1" + local artifact_id="$2" + local version="$3" + cat >> "${POM_FILE}" << EOL + + ${group_id} + ${artifact_id} + ${version} + +EOL +} + +function normalize_beam_version() { + case "${BEAM_VERSION}" in + *-[sS][nN][aA][pP][sS][hH][oO][tT]) + BEAM_VERSION="${BEAM_VERSION%-*}-SNAPSHOT" + ;; + esac +} + function usage() { echo "Usage: $0 [--version ] [--runner ] [--io ] [--list-versions] [--list-ios] [--list-runners] [--debug] [-h|--help]" echo "" @@ -91,6 +125,7 @@ function usage() { echo "" echo "Options:" echo " --version Specify the Apache Beam version (default: ${DEFAULT_BEAM_VERSION})." + echo " SNAPSHOT versions are resolved from Apache's Maven snapshot repository." echo " --runner Specify the Beam runner to use (default: direct)." echo " Supported runners:" echo " direct - DirectRunner (runs locally, good for development)" @@ -208,7 +243,7 @@ function list_runners() { echo "✅ Available runners for Beam ${BEAM_VERSION}:" echo "" - + # Process each runner and provide descriptions while IFS= read -r runner; do case "$runner" in @@ -217,7 +252,7 @@ function list_runners() { echo " Runs locally on your machine. Good for development and testing." ;; "google-cloud-dataflow-java") - echo " dataflow - DataflowRunner" + echo " dataflow - DataflowRunner" echo " Runs on Google Cloud Dataflow for production workloads." ;; flink-*) @@ -273,7 +308,7 @@ function list_runners() { ;; esac done <<< "$runners" - + echo "" echo "💡 Usage: ./beam-sql.sh --runner " echo " Default: direct" @@ -291,7 +326,7 @@ DEBUG_MODE=false while [[ "$#" -gt 0 ]]; do case $1 in - --version) BEAM_VERSION="$2"; shift ;; + --version) BEAM_VERSION="$2"; normalize_beam_version; shift ;; --runner) BEAM_RUNNER=$(echo "$2" | tr '[:upper:]' '[:lower:]'); shift ;; --io) IO_CONNECTORS+=("$2"); shift ;; --list-versions) list_versions; exit 0 ;; @@ -363,7 +398,15 @@ else EOL # Add IO and Runner dependencies for io in "${IO_CONNECTORS[@]}"; do - echo " org.apache.beambeam-sdks-java-io-${io}\${beam.version}" >> "${POM_FILE}" + add_beam_dependency "beam-sdks-java-io-${io}" + case "${io}" in + iceberg) + add_beam_dependency "beam-sdks-java-extensions-sql-iceberg" + add_dependency "org.apache.iceberg" "iceberg-aws" "${ICEBERG_VERSION}" + add_dependency "org.apache.iceberg" "iceberg-aws-bundle" "${ICEBERG_VERSION}" + add_dependency "org.apache.iceberg" "iceberg-gcp" "${ICEBERG_VERSION}" + ;; + esac done RUNNER_ARTIFACT="" case "${BEAM_RUNNER}" in @@ -372,12 +415,25 @@ EOL *) echo "❌ Error: Unsupported runner '${BEAM_RUNNER}'." >&2; exit 1 ;; esac if [ -n "${RUNNER_ARTIFACT}" ]; then - echo " org.apache.beam${RUNNER_ARTIFACT}\${beam.version}" >> "${POM_FILE}" + add_beam_dependency "${RUNNER_ARTIFACT}" fi # Complete the POM with the build section for the maven-shade-plugin cat >> "${POM_FILE}" << EOL + + + apache.snapshots + ${APACHE_SNAPSHOT_REPOSITORY_URL} + + false + + + true + always + + + ${BEAM_VERSION} @@ -395,6 +451,11 @@ cat >> "${POM_FILE}" << EOL + + + true + + diff --git a/sdks/go/examples/wasm/README.md b/sdks/go/examples/wasm/README.md index e4ab54d4a3ed..30fd22f624be 100644 --- a/sdks/go/examples/wasm/README.md +++ b/sdks/go/examples/wasm/README.md @@ -68,13 +68,13 @@ cd $BEAM_HOME Expected output should include the following, from which you acquire the latest flink runner version. ```shell -'flink_versions: 1.17,1.18,1.19,1.20' +'flink_versions: 1.17,1.18,1.19,1.20,2.0,2.1' ``` -#### 2. Set to the latest flink runner version i.e. 1.16 +#### 2. Set to the latest flink runner version i.e. 2.1 ```shell -FLINK_VERSION=1.16 +FLINK_VERSION=2.1 ``` #### 3. In a separate terminal, start the flink runner (It should take a few minutes on the first execution) diff --git a/sdks/go/test/build.gradle b/sdks/go/test/build.gradle index 677134716062..3437ba12f2c7 100644 --- a/sdks/go/test/build.gradle +++ b/sdks/go/test/build.gradle @@ -36,8 +36,10 @@ task dataflowValidatesRunner() { dependsOn ":sdks:go:test:goBuild" doLast { + def pipelineOptions = ["--no_use_public_ips"] def options = [ "--runner dataflow", + "--pipeline_opts \"${pipelineOptions.join(' ')}\"", ] exec { if (fork_java_home) { @@ -59,6 +61,7 @@ task dataflowValidatesRunnerARM64() { doLast { def pipelineOptions = [ // Pipeline options piped directly to Go SDK flags. "--machine_type=t2a-standard-1", + "--no_use_public_ips", ] def options = [ "--runner dataflow", @@ -89,7 +92,7 @@ task flinkValidatesRunner { doFirst { // Copy Flink conf file copy { - from "${project.rootDir}/runners/flink/2.0/src/test/resources/flink-test-config.yaml" + from "${project.rootDir}/runners/flink/${flinkVersion}/src/test/resources/flink-test-config.yaml" into "${project.buildDir}/flink-conf" // Rename the file during the copy process diff --git a/sdks/python/apache_beam/ml/inference/base.py b/sdks/python/apache_beam/ml/inference/base.py index b2441281dd18..f81382bbeecf 100644 --- a/sdks/python/apache_beam/ml/inference/base.py +++ b/sdks/python/apache_beam/ml/inference/base.py @@ -1443,6 +1443,7 @@ def annotations(self): 'model_handler_type': ( f'{self._model_handler.__class__.__module__}' f'.{self._model_handler.__class__.__qualname__}'), + 'model_identifier': self._model_tag, **super().annotations() } @@ -1997,6 +1998,11 @@ def load(): # Ensure the tag we're loading is valid, if not replace it with a valid tag self._cur_tag = self._model_metadata.get_valid_tag(model_tag) if self.use_model_manager: + # Force an import here to avoid missing ModelManager when needed. + # Throw an error if ModelManager is not available since it's required for this code path. + global ModelManager + if ModelManager is None: + from apache_beam.ml.inference.model_manager import ModelManager logging.info("Using Model Manager to manage models automatically.") model_manager = multi_process_shared.MultiProcessShared( lambda: ModelManager(**self._model_manager_args), diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py index c813939d53f1..978a79bf6172 100644 --- a/sdks/python/apache_beam/options/pipeline_options.py +++ b/sdks/python/apache_beam/options/pipeline_options.py @@ -2106,7 +2106,7 @@ def _add_argparse_args(cls, parser): class FlinkRunnerOptions(PipelineOptions): # These should stay in sync with gradle.properties. - PUBLISHED_FLINK_VERSIONS = ['1.17', '1.18', '1.19', '1.20', '2.0'] + PUBLISHED_FLINK_VERSIONS = ['1.17', '1.18', '1.19', '1.20', '2.0', '2.1'] @classmethod def _add_argparse_args(cls, parser): diff --git a/sdks/python/scripts/run_integration_test.sh b/sdks/python/scripts/run_integration_test.sh index b9bb2457c961..4206f9ee9717 100755 --- a/sdks/python/scripts/run_integration_test.sh +++ b/sdks/python/scripts/run_integration_test.sh @@ -42,6 +42,8 @@ # flag is specified, all above flag will be ignored. # Please include all required pipeline options when # using this flag. +# additional_opts -> List of space separated pipeline options. Unlike pipeline_opts, +# it appends to other flags options instead of ignoring them. # # Test related flags: # test_opts -> List of space separated options to configure Pytest test @@ -150,6 +152,11 @@ case $key in shift # past argument shift # past value ;; + --additional_opts) + ADDITIONAL_OPTS="$2" + shift # past argument + shift # past value + ;; --test_opts) TEST_OPTS="$2" shift # past argument @@ -257,7 +264,9 @@ if [[ -z $PIPELINE_OPTS ]]; then fi PIPELINE_OPTS=$(IFS=" " ; echo "${opts[*]}") - + if [[ -n $ADDITIONAL_OPTS ]]; then + PIPELINE_OPTS+=" ${ADDITIONAL_OPTS}" + fi fi # Handle double quotes in PIPELINE_OPTS diff --git a/sdks/python/test-suites/dataflow/common.gradle b/sdks/python/test-suites/dataflow/common.gradle index 27cf2869600c..7c84700e29fa 100644 --- a/sdks/python/test-suites/dataflow/common.gradle +++ b/sdks/python/test-suites/dataflow/common.gradle @@ -245,6 +245,7 @@ task validatesRunnerBatchTests { doLast { def argMap = [ "test_opts" : basicPytestOpts + ["--numprocesses=8"], + "additional_opts": ['--no_use_public_ips'], "sdk_location": project.ext.sdkLocation, "suite" : "validatesRunnerBatchTests-df${pythonVersionSuffix}", "collect": "it_validatesrunner and not no_sickbay_batch" @@ -266,6 +267,7 @@ task validatesRunnerStreamingTests { doFirst { def argMap = [ "test_opts": basicPytestOpts + ["--numprocesses=8"], + "additional_opts": ['--no_use_public_ips'], "streaming": "true", "sdk_location": project.ext.sdkLocation, "suite": "validatesRunnerStreamingTests-df${pythonVersionSuffix}-xdist", diff --git a/sdks/typescript/src/apache_beam/runners/flink.ts b/sdks/typescript/src/apache_beam/runners/flink.ts index 8f80b971da2a..c8f8f57eb080 100644 --- a/sdks/typescript/src/apache_beam/runners/flink.ts +++ b/sdks/typescript/src/apache_beam/runners/flink.ts @@ -28,7 +28,7 @@ import { JavaJarService } from "../utils/service"; const MAGIC_HOST_NAMES = ["[local]", "[auto]"]; // These should stay in sync with gradle.properties. -const PUBLISHED_FLINK_VERSIONS = ["1.17", "1.18", "1.19", "1.20", "2.0"]; +const PUBLISHED_FLINK_VERSIONS = ["1.17", "1.18", "1.19", "1.20", "2.0", "2.1"]; const defaultOptions = { flinkMaster: "[local]",