This is an automated email from the ASF dual-hosted git repository. Abacn pushed a commit to branch restore-load in repository https://gitbox.apache.org/repos/asf/beam.git
commit 6e148dfe1d2eb12815967d59f7ab851983334c36 Author: Yi Hu <[email protected]> AuthorDate: Thu Aug 13 15:34:14 2026 -0400 debug only --- .../beam_LoadTests_Go_GBK_Flink_Batch.yml | 18 +++++++-------- .../go_GBK_Flink_Batch_100b.txt | 4 +++- .../go_GBK_Flink_Batch_100kb.txt | 4 +++- .../go_GBK_Flink_Batch_10b.txt | 4 +++- .../go_GBK_Flink_Batch_Fanout_4.txt | 4 +++- .../go_GBK_Flink_Batch_Fanout_8.txt | 4 +++- .../go_GBK_Flink_Batch_Reiteration_10KB.txt | 4 +++- .test-infra/dataproc/flink_cluster.sh | 6 +++++ .../flink.go => options/flinkopts/options.go} | 26 ++++++++-------------- sdks/go/pkg/beam/runner.go | 1 + sdks/go/pkg/beam/runners/flink/flink.go | 2 ++ .../go/pkg/beam/runners/universal/runnerlib/job.go | 10 +++++++++ sdks/go/pkg/beam/runners/universal/universal.go | 1 + sdks/go/pkg/beam/x/beamx/run.go | 9 ++++++-- 14 files changed, 62 insertions(+), 35 deletions(-) diff --git a/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml b/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml index 731c5b21274..66ef991a3c5 100644 --- a/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml +++ b/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml @@ -48,13 +48,13 @@ env: INFLUXDB_USER: ${{ secrets.INFLUXDB_USER }} INFLUXDB_USER_PASSWORD: ${{ secrets.INFLUXDB_USER_PASSWORD }} GCLOUD_REGION: us-central1 - CLUSTER_NAME: beam-loadtests-go-gbk-flink-batch-${{ github.run_id }} + CLUSTER_NAME: beam-loadtests-go-sideinput-flink-batch-31733216613 GCS_BUCKET: gs://beam-flink-cluster FLINK_TASKMANAGER_SLOTS: 2 DETACHED_MODE: true HARNESS_IMAGES_TO_PULL: gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest JOB_SERVER_IMAGE: gcr.io/apache-beam-testing/beam_portability/beam_flink_job_server:latest-flink2.2 - ARTIFACTS_DIR: gs://beam-flink-cluster/beam-loadtests-go-gbk-flink-batch-${{ github.run_id }} + ARTIFACTS_DIR: gs://beam-flink-cluster/beam-loadtests-go-sideinput-flink-batch-31733216613 jobs: beam_LoadTests_Go_GBK_Flink_Batch: @@ -97,11 +97,9 @@ jobs: ${{ github.workspace }}/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt arguments: | --job_name=load-tests-go-flink-batch-gbk-$(date '+%m%d%H%M%S' --utc) - - name: Start Flink with parallelism 5 - env: - FLINK_NUM_WORKERS: 5 + - name: Connect to existing Flink cluster run: | - cd ${{ github.workspace }}/.test-infra/dataproc; ./flink_cluster.sh create + cd ${{ github.workspace }}/.test-infra/dataproc; ./flink_cluster.sh tunnel # The env variables are created and populated in the test-arguments-action as "<github.job>_test_arguments_<argument_file_paths_index>" - name: run GBK Flink Batch Go Load Test 1 (10 b records) timeout-minutes: 120 @@ -162,7 +160,7 @@ jobs: -PloadTest.mainClass=group_by_key \ -Prunner=FlinkRunner \ '-PloadTest.args=${{ env.beam_LoadTests_Go_GBK_Flink_Batch_test_arguments_5 }}' \ - - name: Teardown Flink - if: always() - run: | - ${{ github.workspace }}/.test-infra/dataproc/flink_cluster.sh delete + # - name: Teardown Flink + # if: always() + # run: | + # ${{ github.workspace }}/.test-infra/dataproc/flink_cluster.sh delete diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100b.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100b.txt index d5c1d865e18..a815510613d 100644 --- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100b.txt +++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100b.txt @@ -23,4 +23,6 @@ --endpoint=localhost:8099 --environment_type=DOCKER --environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest ---runner=FlinkRunner \ No newline at end of file +--runner=FlinkRunner +--state_backend=rocksdb +--state_backend_storage_path=file:///tmp/flink-state \ No newline at end of file diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100kb.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100kb.txt index a188f8c0978..e426bc7738e 100644 --- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100kb.txt +++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100kb.txt @@ -23,4 +23,6 @@ --endpoint=localhost:8099 --environment_type=DOCKER --environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest ---runner=FlinkRunner \ No newline at end of file +--runner=FlinkRunner +--state_backend=rocksdb +--state_backend_storage_path=file:///tmp/flink-state \ No newline at end of file diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_10b.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_10b.txt index 0bb35b26436..7cc5c328472 100644 --- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_10b.txt +++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_10b.txt @@ -23,4 +23,6 @@ --endpoint=localhost:8099 --environment_type=DOCKER --environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest ---runner=FlinkRunner \ No newline at end of file +--runner=FlinkRunner +--state_backend=rocksdb +--state_backend_storage_path=file:///tmp/flink-state \ No newline at end of file diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_4.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_4.txt index 4378d56a8f8..335418079aa 100644 --- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_4.txt +++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_4.txt @@ -23,4 +23,6 @@ --endpoint=localhost:8099 --environment_type=DOCKER --environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest ---runner=FlinkRunner \ No newline at end of file +--runner=FlinkRunner +--state_backend=rocksdb +--state_backend_storage_path=file:///tmp/flink-state \ No newline at end of file diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_8.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_8.txt index 43292d57717..3d1fe0930d0 100644 --- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_8.txt +++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_8.txt @@ -23,4 +23,6 @@ --endpoint=localhost:8099 --environment_type=DOCKER --environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest ---runner=FlinkRunner \ No newline at end of file +--runner=FlinkRunner +--state_backend=rocksdb +--state_backend_storage_path=file:///tmp/flink-state \ No newline at end of file diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt index c4d33c21482..59672c2217f 100644 --- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt +++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt @@ -23,4 +23,6 @@ --endpoint=localhost:8099 --environment_type=DOCKER --environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest ---runner=FlinkRunner \ No newline at end of file +--runner=FlinkRunner +--state_backend=rocksdb +--state_backend_storage_path=file:///tmp/flink-state \ No newline at end of file diff --git a/.test-infra/dataproc/flink_cluster.sh b/.test-infra/dataproc/flink_cluster.sh index e146aa4b6ce..17180ced89d 100755 --- a/.test-infra/dataproc/flink_cluster.sh +++ b/.test-infra/dataproc/flink_cluster.sh @@ -183,6 +183,12 @@ function create() { start_tunnel } +# Connects to an existing Flink cluster and opens an SSH tunnel. +function tunnel() { + get_leader + start_tunnel +} + # Resizes an active Flink cluster. function resize() { if [[ -z "$FLINK_NUM_WORKERS" ]]; then diff --git a/sdks/go/pkg/beam/runners/flink/flink.go b/sdks/go/pkg/beam/options/flinkopts/options.go similarity index 57% copy from sdks/go/pkg/beam/runners/flink/flink.go copy to sdks/go/pkg/beam/options/flinkopts/options.go index c79c778a778..c4fd6e14400 100644 --- a/sdks/go/pkg/beam/runners/flink/flink.go +++ b/sdks/go/pkg/beam/options/flinkopts/options.go @@ -13,23 +13,15 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package flink contains the Flink runner. -package flink +// Package flinkopts contains pipeline options for the Apache Flink runner. +package flinkopts -import ( - "context" +import "flag" - "github.com/apache/beam/sdks/v2/go/pkg/beam" - "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/universal" -) - -func init() { - beam.RegisterRunner("flink", Execute) - beam.RegisterRunner("FlinkRunner", Execute) -} +var ( + // StateBackend is the state backend to use for Flink (e.g. 'rocksdb', 'hashmap', 'filesystem'). + StateBackend = flag.String("state_backend", "", "The state backend to use for Flink (optional, e.g. 'rocksdb', 'hashmap', 'filesystem').") -// Execute runs the given pipeline on Flink. Convenience wrapper over the -// universal runner. -func Execute(ctx context.Context, p *beam.Pipeline) (beam.PipelineResult, error) { - return universal.Execute(ctx, p) -} + // StateBackendStoragePath is the storage path for the Flink state backend (e.g. 'file:///tmp/flink-state', 'gs://...'). + StateBackendStoragePath = flag.String("state_backend_storage_path", "", "Storage path for the Flink state backend.") +) diff --git a/sdks/go/pkg/beam/runner.go b/sdks/go/pkg/beam/runner.go index c9747da602e..d3557b24f1a 100644 --- a/sdks/go/pkg/beam/runner.go +++ b/sdks/go/pkg/beam/runner.go @@ -39,6 +39,7 @@ func RegisterRunner(name string, fn func(ctx context.Context, p *Pipeline) (Pipe // to define a "runner" with no default as a flag to let users control runner // selection. func Run(ctx context.Context, runner string, p *Pipeline) (PipelineResult, error) { + log.Infof(ctx, "Executing pipeline with runner: %v", runner) fn, ok := runners[runner] if !ok { log.Exitf(ctx, "Runner %v not registered. Forgot to _ import it?", runner) diff --git a/sdks/go/pkg/beam/runners/flink/flink.go b/sdks/go/pkg/beam/runners/flink/flink.go index c79c778a778..11311ca76ce 100644 --- a/sdks/go/pkg/beam/runners/flink/flink.go +++ b/sdks/go/pkg/beam/runners/flink/flink.go @@ -20,6 +20,8 @@ import ( "context" "github.com/apache/beam/sdks/v2/go/pkg/beam" + // Import flinkopts for the side effect of registering Flink-specific command-line flags. + _ "github.com/apache/beam/sdks/v2/go/pkg/beam/options/flinkopts" "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/universal" ) diff --git a/sdks/go/pkg/beam/runners/universal/runnerlib/job.go b/sdks/go/pkg/beam/runners/universal/runnerlib/job.go index 81ff5a5eb94..81259867f18 100644 --- a/sdks/go/pkg/beam/runners/universal/runnerlib/job.go +++ b/sdks/go/pkg/beam/runners/universal/runnerlib/job.go @@ -29,6 +29,7 @@ import ( "github.com/apache/beam/sdks/v2/go/pkg/beam/log" jobpb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/jobmanagement_v1" pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1" + "google.golang.org/protobuf/types/known/structpb" ) // JobOptions capture the various options for submitting jobs @@ -69,6 +70,15 @@ func Prepare(ctx context.Context, client jobpb.JobServiceClient, p *pipepb.Pipel if err != nil { return "", "", "", errors.WithContext(err, "producing pipeline options") } + if options.Fields == nil { + options.Fields = make(map[string]*structpb.Value) + } + for k, v := range raw.Options.Options { + key := fmt.Sprintf("beam:option:%v:v1", k) + if _, exists := options.Fields[key]; !exists && v != "" { + options.Fields[key] = structpb.NewStringValue(v) + } + } req := &jobpb.PrepareJobRequest{ Pipeline: p, PipelineOptions: options, diff --git a/sdks/go/pkg/beam/runners/universal/universal.go b/sdks/go/pkg/beam/runners/universal/universal.go index 25325b8fe9c..36a3f45a8aa 100644 --- a/sdks/go/pkg/beam/runners/universal/universal.go +++ b/sdks/go/pkg/beam/runners/universal/universal.go @@ -60,6 +60,7 @@ func Execute(ctx context.Context, p *beam.Pipeline) (beam.PipelineResult, error) if err != nil { return nil, err } + log.Infof(ctx, "Universal runner connecting to endpoint: %v", endpoint) edges, _, err := p.Build() if err != nil { diff --git a/sdks/go/pkg/beam/x/beamx/run.go b/sdks/go/pkg/beam/x/beamx/run.go index ff3583917b7..885728ab230 100644 --- a/sdks/go/pkg/beam/x/beamx/run.go +++ b/sdks/go/pkg/beam/x/beamx/run.go @@ -20,6 +20,7 @@ import ( "context" "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/log" "github.com/apache/beam/sdks/v2/go/pkg/beam/runners" // common runner flag. // Import the reflection-optimized runtime. @@ -54,7 +55,9 @@ func getRunner() string { // defaults to the prism runner, but all beam-distributed runners and textio // filesystems are implicitly registered. func Run(ctx context.Context, p *beam.Pipeline) error { - _, err := beam.Run(ctx, getRunner(), p) + r := getRunner() + log.Infof(ctx, "beamx: using runner %q (flag value %q)", r, *runner) + _, err := beam.Run(ctx, r, p) return err } @@ -62,5 +65,7 @@ func Run(ctx context.Context, p *beam.Pipeline) error { // flag "runner". Returns a beam.PipelineResult objects, which can be // accessed to query the pipeline's metrics. func RunWithMetrics(ctx context.Context, p *beam.Pipeline) (beam.PipelineResult, error) { - return beam.Run(ctx, getRunner(), p) + r := getRunner() + log.Infof(ctx, "beamx: using runner %q (flag value %q)", r, *runner) + return beam.Run(ctx, r, p) }
