gemini-code-assist[bot] commented on code in PR #38411:
URL: https://github.com/apache/beam/pull/38411#discussion_r3204865024
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java:
##########
@@ -1245,7 +1245,7 @@ public DataflowPipelineJob run(Pipeline pipeline) {
// to Runner v2.
if (DataflowRunner.isMultiLanguagePipeline(pipeline) ||
includesTransformUpgrades(pipeline)) {
List<String> experiments = firstNonNull(options.getExperiments(),
Collections.emptyList());
- if (!experiments.contains("use_runner_v2")) {
+ if (!experiments.contains("use_runner_v2") ||
!experiments.contains("enable_portable_runner")) {
Review Comment:
The logic here is incorrect. Using `||` with negated conditions (`!A || !B`)
is equivalent to `!(A && B)`, which means this block will execute if *either*
experiment is missing. If a user has already provided `use_runner_v2`, they
have explicitly enabled Runner v2, but this condition will still be true
(because `enable_portable_runner` is missing), causing a misleading log message
about "Automatically enabling" the runner. It should use `&&` to only trigger
if *neither* experiment is present.
```suggestion
if (!experiments.contains("use_runner_v2") &&
!experiments.contains("enable_portable_runner")) {
```
##########
sdks/go/pkg/beam/runners/dataflow/dataflow.go:
##########
@@ -349,7 +349,10 @@ func getJobOptions(ctx context.Context, streaming bool)
(*dataflowlib.JobOptions
if strings.Contains(e, "use_portable_job_submission") {
portaSubmission = true
}
- if strings.Contains(e, "disable_runner_v2") ||
strings.Contains(e, "disable_runner_v2_until_2023") || strings.Contains(e,
"disable_prime_runner_v2") {
+ if strings.Contains(e, "enable_portable_runner") {
+ portableRunnerSet = true
+ }
+ if strings.Contains(e, "disable_runner_v2") ||
strings.Contains(e, "disable_runner_v2_until_2023") || strings.Contains(e,
"disable_prime_runner_v2") || strings.Contains(e, "disable_portable_runner") ||
strings.Contains(e, "enable_java_streaming_runner") {
return nil, errors.New("detected one of the following
experiments: disable_runner_v2 | disable_runner_v2_until_2023 |
disable_prime_runner_v2. Disabling runner v2 is no longer supported as of Beam
version 2.45.0+")
Review Comment:
The error message should be updated to include the new experiments
(`disable_portable_runner` and `enable_streaming_java_runner`) that now trigger
this validation error.
```suggestion
return nil, errors.New("detected one of the following
experiments: disable_runner_v2 | disable_runner_v2_until_2023 |
disable_prime_runner_v2 | disable_portable_runner |
enable_streaming_java_runner. Disabling runner v2 is no longer supported as of
Beam version 2.45.0+")
```
##########
sdks/go/pkg/beam/runners/dataflow/dataflow.go:
##########
@@ -349,7 +349,10 @@ func getJobOptions(ctx context.Context, streaming bool)
(*dataflowlib.JobOptions
if strings.Contains(e, "use_portable_job_submission") {
portaSubmission = true
}
- if strings.Contains(e, "disable_runner_v2") ||
strings.Contains(e, "disable_runner_v2_until_2023") || strings.Contains(e,
"disable_prime_runner_v2") {
+ if strings.Contains(e, "enable_portable_runner") {
+ portableRunnerSet = true
+ }
+ if strings.Contains(e, "disable_runner_v2") ||
strings.Contains(e, "disable_runner_v2_until_2023") || strings.Contains(e,
"disable_prime_runner_v2") || strings.Contains(e, "disable_portable_runner") ||
strings.Contains(e, "enable_java_streaming_runner") {
Review Comment:
There is a typo in the experiment name: `enable_java_streaming_runner`
should be `enable_streaming_java_runner` to maintain consistency with the Java
and Python SDKs and the PR description. This is also evidenced by the test case
in `dataflow_test.go` which uses the correct name.
```suggestion
if strings.Contains(e, "disable_runner_v2") ||
strings.Contains(e, "disable_runner_v2_until_2023") || strings.Contains(e,
"disable_prime_runner_v2") || strings.Contains(e, "disable_portable_runner") ||
strings.Contains(e, "enable_streaming_java_runner") {
```
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java:
##########
@@ -1256,7 +1256,9 @@ public DataflowPipelineJob run(Pipeline pipeline) {
if (useUnifiedWorker(options)) {
if (hasExperiment(options, "disable_runner_v2")
|| hasExperiment(options, "disable_runner_v2_until_2023")
- || hasExperiment(options, "disable_prime_runner_v2")) {
+ || hasExperiment(options, "disable_prime_runner_v2")
+ || hasExperiment(options, "disable_portable_runner")
+ || hasExperiment(options, "enable_streaming_java_runner")) {
throw new IllegalArgumentException(
"Runner V2 both disabled and enabled: at least one of
['beam_fn_api', 'use_unified_worker', 'use_runner_v2',
'use_portable_job_submission'] is set and also one of ['disable_runner_v2',
'disable_runner_v2_until_2023', 'disable_prime_runner_v2'] is set.");
Review Comment:
The error message is outdated and should be updated to include the newly
added experiments (`enable_portable_runner`, `disable_portable_runner`, and
`enable_streaming_java_runner`) to accurately reflect the conditions that
trigger this exception.
```suggestion
"Runner V2 both disabled and enabled: at least one of
['beam_fn_api', 'use_unified_worker', 'use_runner_v2',
'use_portable_job_submission', 'enable_portable_runner'] is set and also one of
['disable_runner_v2', 'disable_runner_v2_until_2023',
'disable_prime_runner_v2', 'disable_portable_runner',
'enable_streaming_java_runner'] is set.");
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
