Abacn commented on PR #39735:
URL: https://github.com/apache/beam/pull/39735#issuecomment-5269612451
> @johnjcasey @Abacn
>
> I think this is fine - but is there some nuance where it simply won't
work? (the runner should really reject it, not the transform itself, but I know
that KafkaIO does a bunch of logic around various flags and whatnot)
>
> Is there a postsubmit with trigger file we should add here?
Agree, it's no longer needed. AI investigation (checked manually):
This check is outdated, inaccurate, and should be removed or updated.
1. Inaccurate for Dataflow Runner v2 users: In DataflowRunner.java, the
portable / Runner v2 engine (useUnifiedWorker) can be enabled by any of several
experiment flags:
```java
static boolean useUnifiedWorker(DataflowPipelineOptions options) {
return hasExperiment(options, "beam_fn_api")
|| hasExperiment(options, "use_runner_v2")
|| hasExperiment(options, "use_unified_worker")
|| hasExperiment(options, "use_portable_job_submission")
|| hasExperiment(options, "enable_portable_runner");
}
```
If a user launches a Dataflow pipeline with --experiments=use_runner_v2 (the
standard and recommended flag for Runner v2) without also explicitly passing
--experiments=beam_fn_api, the check in KafkaIO.java:L1663-1666 will throw an
error:
```
IllegalArgumentException: Kafka Dynamic Read requires enabling experiment
beam_fn_api.
```
even though the job is running on Runner v2.
2. Historical artifact from 2021: Git history (commit 45a2193e4f7) shows
that this check was introduced in March 2021 when SDF support in Beam Java SDK
was first being rolled out and required explicit activation via "beam_fn_api".
3. Already handled by the SDK implementation compatibility matrix: KafkaIO
now uses KafkaIOReadImplementationCompatibility.java, where DYNAMIC_READ is
properly mapped exclusively to the SDF implementation (ReadFromKafkaViaSDF).
Hardcoding a runner-specific experiment string inside KafkaIO.expand()
breaks runner-agnostic SDK architecture. If a runner check is desired for
Dataflow specifically, it should either rely on
DataflowRunner.useUnifiedWorker(options) or let the runner transform override
handle capability validation.
--
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]
