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
The following commit(s) were added to refs/heads/restore-load by this push:
new bfe820832a0 debug only
bfe820832a0 is described below
commit bfe820832a08bf5209c19f5512563e5fef39cc34
Author: Yi Hu <[email protected]>
AuthorDate: Thu Aug 13 15:34:14 2026 -0400
debug only
---
.github/workflows/beam_LoadTests_Go_CoGBK_Flink_batch.yml | 8 ++++----
sdks/go/pkg/beam/runner.go | 1 +
sdks/go/pkg/beam/runners/universal/universal.go | 1 +
sdks/go/pkg/beam/x/beamx/run.go | 9 +++++++--
4 files changed, 13 insertions(+), 6 deletions(-)
diff --git a/.github/workflows/beam_LoadTests_Go_CoGBK_Flink_batch.yml
b/.github/workflows/beam_LoadTests_Go_CoGBK_Flink_batch.yml
index f59ad9900e5..ce2119b98a2 100644
--- a/.github/workflows/beam_LoadTests_Go_CoGBK_Flink_batch.yml
+++ b/.github/workflows/beam_LoadTests_Go_CoGBK_Flink_batch.yml
@@ -127,7 +127,7 @@ jobs:
-PloadTest.mainClass=cogbk \
-Prunner=FlinkRunner \
'-PloadTest.args=${{
env.beam_LoadTests_Go_CoGBK_Flink_Batch_test_arguments_3 }}'
- - 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/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/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)
}
