You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2026/08/13 22:14:34 UTC
(beam) branch restore-load updated (452a4abd48e -> bdb59472e7d)
This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a change to branch restore-load
in repository https://gitbox.apache.org/repos/asf/beam.git
discard 452a4abd48e debug only
new bdb59472e7d debug only
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (452a4abd48e)
\
N -- N -- N refs/heads/restore-load (bdb59472e7d)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../load-tests-pipeline-options/go_GBK_Flink_Batch_100b.txt | 4 +++-
.../load-tests-pipeline-options/go_GBK_Flink_Batch_100kb.txt | 4 +++-
.../load-tests-pipeline-options/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 +++-
sdks/go/pkg/beam/runners/flink/flink.go | 2 ++
sdks/go/pkg/beam/runners/universal/runnerlib/job.go | 10 ++++++++++
8 files changed, 30 insertions(+), 6 deletions(-)
(beam) 01/01: debug only
Posted by yh...@apache.org.
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 bdb59472e7d0bc0b7b2f7b46367d074ca5a37f9b
Author: Yi Hu <ya...@google.com>
AuthorDate: Thu Aug 13 15:34:14 2026 -0400
debug only
---
.../workflows/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 ++++++
sdks/go/pkg/beam/runner.go | 1 +
sdks/go/pkg/beam/runners/flink/flink.go | 2 ++
sdks/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 +++++++--
13 files changed, 53 insertions(+), 18 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/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)
}