You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "damnMeddlingKid (via GitHub)" <gi...@apache.org> on 2024/03/02 04:50:21 UTC
[I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
damnMeddlingKid opened a new issue, #30469:
URL: https://github.com/apache/beam/issues/30469
### What happened?
## Issue
When running a pipeline with KafkaIO and withDynamicReads enabled I get the following error in Dataflow on Runner V2
```
Exception in thread "main" java.lang.IllegalStateException:
The current runner does not support SDF-based Kafka read properly and the replacement runner lacks the support for the following properties: [DYNAMIC_READ, WATCH_TOPIC_PARTITION_DURATION].
For example if you are using Dataflow then consider using Dataflow Runner v2.; see consecutive INFO logs for details.
```
This is despite setting the `use_runner_v2` and `beam_fn_api` experiment flags.
Steps to reproduce:
1) Create a pipeline with KafkaIO and dynamic reads, heres my configuration
```
KafkaIO
.read[Array[Byte], Array[Byte]]
.withBootstrapServers(kafkaBootstrapServers)
.withDynamicRead(Duration.standardMinutes(15))
.withTopics(kafkaTopics.asJava)
.withReadCommitted()
.withConsumerConfigUpdates(kafkaConsumerProperties.toMap[String, AnyRef].asJava)
.commitOffsetsInFinalize
.withKeyDeserializer(classOf[ByteArrayDeserializer])
.withValueDeserializer(classOf[ByteArrayDeserializer])
```
2) Enable experiment flags for runner v2 and beam function api: `use_runner_v2`, `beam_fn_api`
3) Run the pipeline in Dataflow
## Investigation
I believe this error is happening because beam always [replaces V1 transforms](https://github.com/apache/beam/blob/935ca9c58d61e5d070056d36c21c6b6c03a97908/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java#L1227C1-L1229C35) and the transform replacement for V1 is a non-SDF based reader that doesn't support dynamic reads.
so when the replacement is attempted [Beam correctly validates](https://github.com/apache/beam/blob/master/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L1539-L1543) that the replacement is not compatible with Dynamic Reads and throws a compatibility exception.
I would like to propose that we only make this replacement for runner v1 like so https://github.com/damnMeddlingKid/beam/pull/1/files
```
if (!hasExperiment(options, "use_runner_v2")) {
overridesBuilder.add(KafkaIO.Read.KAFKA_READ_OVERRIDE);
}
```
### Issue Priority
Priority: 3 (minor)
### Issue Components
- [ ] Component: Python SDK
- [X] 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
- [X] Component: Google Cloud Dataflow Runner
--
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: github-unsubscribe@beam.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on issue #30469:
URL: https://github.com/apache/beam/issues/30469#issuecomment-1986184429
Dataflow v2 doesn't use the translation that occurs on the client so just FYI that's a red herring. The client-side translation exists only to serve v1. So the "create job" request always sends the request through the Dataflow API client for v1 and uploads the original Beam proto for v2 to translate later in its own way.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles closed issue #30469: [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2
URL: https://github.com/apache/beam/issues/30469
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on issue #30469:
URL: https://github.com/apache/beam/issues/30469#issuecomment-1986199984
Really unusual that we didn't hit this before. The codepaths are clearly internally inconsistent. You can't do a dual (v1 and v2 combined) submission when one of the codepaths doesn't even have a way to translate/implement (the v1 translation will fail or just be totally wrong).
I think what we want to do if we find features used that make v1 unusable is to truly just zero out the whole v1 submission. This should happen in the DataflowRunner not in KafkaIO. Your change to just not do the override is a decent mitigation since I don't think we have a good way otherwise. But just having some `V1UnsupportedFeatureException(String)` and then catching that at the top level and setting the v1beta3 `workflow` to null would be the ultimate goal. Not sure if that will confuse downstream stuff though...
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on issue #30469:
URL: https://github.com/apache/beam/issues/30469#issuecomment-1992026050
Just waiting for some long-running tests to merge the fix.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on issue #30469:
URL: https://github.com/apache/beam/issues/30469#issuecomment-1986186256
It is an explicit goal that nothing on the client side observes the `use_runner_v2` experiment flag, to ensure this works right. We aren't all the way there. Perhaps this is a bug in validation actually.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "AnandInguva (via GitHub)" <gi...@apache.org>.
AnandInguva commented on issue #30469:
URL: https://github.com/apache/beam/issues/30469#issuecomment-1992021675
+1 hitting this issue today on 2.54.0
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on issue #30469:
URL: https://github.com/apache/beam/issues/30469#issuecomment-1986080378
This is an odd scenario, because it doesn't appear to have come up in testing. That suggests there is either something different about your env, or there is a gap in our tests.
I think your change is worth attempting, but I wonder if it will break the double-submission model of the dataflow runner
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "damnMeddlingKid (via GitHub)" <gi...@apache.org>.
damnMeddlingKid commented on issue #30469:
URL: https://github.com/apache/beam/issues/30469#issuecomment-1992753372
@AnandInguva was this working for you in the previous version of beam ?.
wondering when the regression happened.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on issue #30469:
URL: https://github.com/apache/beam/issues/30469#issuecomment-1997771526
I think it must have been #14633. That was many years ago, though, so if it just happened recently then it would have to be something else.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: KafkaIO cannot enable Dynamic Reads on Dataflow Runner V2 [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on issue #30469:
URL: https://github.com/apache/beam/issues/30469#issuecomment-1989563258
PR looks green but running a few more tests. I didn't add a specific regression test for this one, which I suppose I ought to.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org