[BEAM-10670] Use non-SDF based translation for Read by default on all runners except Dataflow

[boyuanzz]

Applied the same logic as #14755 except Dataflow

R: @kennknowles@iemejia
cc: @ibzib@lukecwik

Please add a meaningful description for your change here

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

ValidatesRunner compliance status (on master branch)

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---			---		---
Java
Python	---			---		---
XLang				---		---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---			---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[kennknowles]

@xinyuiscool is this issue something that affects you or that you have noticed?

[kennknowles]

@pulasthi is this issue something that affects you or that you have noticed?

[boyuanzz]

Unfortunately, adding io-kafka into direct-runner will cause circular dependency.

[iemejia]

You may solve the circular dependency issue and the need for the extra dependency on runners by giving the specific Kafka transform a stable URN and then do the translation override via URNs (see for example how they do for PTransformTranslation.READ_TRANSFORM_URN).

[boyuanzz]

READ_TRANSFORM_URN

I'm hesitated to do so because ReadFromKafkaViaSDF should just be a normal DoFn. One way I'm thinking about is to always populate beam_fn_api_use_deprecated_read and use_deprecated_read when use_sdf_read is not specified. beam_fn_api_use_deprecated_read and use_deprecated_read control the KafkaIO expansion directly.

[iemejia]

Maybe worth to add the TODOs to remove this once SDF is fixed because it does not make sense that runners depend on IOs at least not for non test scopes

[boyuanzz]

Good point! Added TODO all build files.

[ihji]

One way I'm thinking about is to always populate beam_fn_api_use_deprecated_read and use_deprecated_read when use_sdf_read is not specified.

I think the method mentioned above might be a better way to do this since we could avoid the io dependency on runners.

[boyuanzz]

Usually populating experiments from runner is the least preferred one for me. With applying override utils from transform directly, we will have a centralized place to change the logic if needed.

[ihji]

Yeah, it makes sense. But I still prefer to not introduce the additional io dependency to runners given that this change is temporary and will be reverted soon once we fix the performance issue (I'm not totally against the current implementation though so it's good to go if other folks don't have any preference).

[iemejia]

I really want this to move on but I have to say that I agree with @ihji, another reason is that adding the KafkaIO dependency to almost all runners will bring a lot of the extra baggage that KafkaIO has including extra confluent and spring dependencies that will leak into the user system's classpath and can produce conflicts.

[boyuanzz]

Changed to populating experiments directly.

[boyuanzz]

@ihji Heejong, would you like to take another look?

[ihji]

LGTM. Thanks!

[ihji]

A side question: does this mean that portable Flink doesn't currently support "use_deprecated_read" flag? What happens if we specify "use_deprecated_read" for portable Flink? Runtime error?

[boyuanzz]

There is no deprecated read under beam_fn_api context. Splittable DoFn is the only source-like operation when with beam_fn_api. So if a fnapi job is using deprecated read, there is no runtime error but the pipeline will just stuck.

[boyuanzz]

But good part is that I don't think we have many fnapi usages on OSS runners.

[boyuanzz]

I'll open a cherry-pick as well.

[iemejia]

Thanks @boyuanzz for working on this!

[ibzib]

I think this broke Spark postcommits: BEAM-12374

[boyuanzz]

I think this broke Spark postcommits: BEAM-12374

Yeah I'm looking into it now and I should have a fix today.

[boyuanzz]

Sorry that I misunderstood the execution of Transform.expand(). The transform expansion happens during applying a PTransform to a certain PCollection before we calling runner.run(pipeline). So populating experiments doesn't work for our need. Another way I'm thinking about is to have KafkaIO run expansion based on the given runner. It can be done via checking runner class prefix.