[Bug]: Dataflow worker container resolved to legacy runner label if not explicitly disable/enable runner v2 in 2.54.0+.dev

[Abacn]

What happened?

Run a batch job on Dataflow runner use SNAPSHOT version, the Dataflow job now stucks, with error

Container image gcr.io/cloud-dataflow/v1beta3/beam_java8_sdk:beam-master-20240306 not downloaded yet.

This is seen in #30505 (comment)

It does not affect release as both dataflowLegacyContainerVersion and dataflowFnapiContainerVersion are identical

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[Abacn]

The logic of determining contaner label is here:

beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

Line 2516 in ef919e2

static String getDefaultContainerImageUrl(DataflowPipelineOptions options) {

calls

beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

Line 2542 in ef919e2

static String getDefaultContainerVersion(DataflowPipelineOptions options) {

calls

beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

Line 2554 in ef919e2

static boolean useUnifiedWorker(DataflowPipelineOptions options) {

As for batch pipeline, it should still resolve to `gcr.io/cloud-dataflow/v1beta3/beam-java-batch:beam-master-20240306'. Does the replace happen on the Dataflow side ?

[Abacn]

This also affects

beam/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncher.java

Lines 261 to 267 in a3e5ac8

	String runner = LEGACY_RUNNER;
	Environment environment = job.getEnvironment();
	if (environment != null
	&& environment.getExperiments() != null
	&& environment.getExperiments().contains("use_runner_v2")) {
	runner = RUNNER_V2;
	}

the throughput metrics are not correctly reported in IOLoadTests for read pipelines