[BEAM-11325] ReadFromKafkaDoFn should stop reading when topic/partition is removed or marked as stopped #13710

boyuanzz · 2021-01-09T00:11:56Z

For design details, please refer to the doc here: https://docs.google.com/document/d/1FU3GxVRetHPLVizP3Mdv6mP5tpjZ3fd99qNjUI5DT5k/edit#

Next step is to add support of starting reading from new added TopicPartitions.

r: @dpmills @aromanenko-dev

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.

Post-Commit Tests Status (on master branch)

Lang	Dataflow	Samza	Twister2
Go	---	---	---
Java
Python		---	---
XLang		---	---

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.

aromanenko-dev · 2021-01-15T17:08:28Z

Thanks, I'll take a look!

aromanenko-dev

LGTM in general, I left several comments, ptal.

Also, please, add Jira ID to commit message as a prefix.

Don't you think it will make sense to add a IT that test this functionality against "real" (see KafkaIOIT) Kafka cluster as well?

aromanenko-dev · 2021-01-19T15:41:00Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

+     * A custom {@link SerializableFunction} that determines whether the {@link ReadFromKafkaDoFn}
+     * should stop reading from the given {@link TopicPartition}.
+     */
+    public ReadSourceDescriptors<K, V> withCheckStopReadingFn(


Could you add an example to the main KafkaIO Javadoc how to use this functionality to make it more clear for users?

Yeah, will do. I'll also add support to commit transform together in this PR for reverting safety.

I added more related javadoc around KafkaIO.

Thinking about the commit transform again. I don't think additional changes required here because:

For stopped TopicPartition, we still want to commit the offset

For deleted TopicPartition, we should get exceptions from Kafka Consumer API, which has been handled in the Commit transform.

I'll merge this PR if all these look good to you : )

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java

aromanenko-dev · 2021-01-19T16:52:49Z

Run Java KafkaIO Performance Test

boyuanzz · 2021-01-19T19:11:34Z

The IT failed on pulling sdk harness docker image.

boyuanzz · 2021-01-19T19:11:40Z

Run Java KafkaIO Performance Test

boyuanzz · 2021-01-19T19:51:20Z

Don't you think it will make sense to add a IT that test this functionality against "real" (see KafkaIOIT) Kafka cluster as well?

I think ReadFromKafkaDoFnTest gives a good coverage and it may not be easy to add such an IT within current framework(it might be easier with Kafka Admin API but I need to have a try on that). I would prefer to have ITs in one separated PR later. What do you think?

boyuanzz · 2021-01-19T22:06:54Z

It seems like KafkaIOIT has been failing for more than 1 month. The failure is the pipeline fails to pull java sdk harness docker image from gcr. It looks like an issue of test configuration.

…on is removed or marked as stopped.

aromanenko-dev · 2021-01-20T15:56:02Z

@boyuanzz

I would prefer to have ITs in one separated PR later. What do you think?

Sounds good to me.

aromanenko-dev · 2021-01-21T17:10:24Z

Run Java KafkaIO Performance Test

aromanenko-dev · 2021-01-21T17:18:52Z

It seems like KafkaIOIT has been failing for more than 1 month. The failure is the pipeline fails to pull java sdk harness docker image from gcr. It looks like an issue of test configuration.

Do you have an idea why it got broken and how to fix it? This IT is very valuable since it runs tests agains different Kafka versions and, therefore, test the compatibility.

Other than that, I think we can merge this PR.

boyuanzz · 2021-01-21T18:22:27Z

It seems like KafkaIOIT has been failing for more than 1 month. The failure is the pipeline fails to pull java sdk harness docker image from gcr. It looks like an issue of test configuration.

Do you have an idea why it got broken and how to fix it? This IT is very valuable since it runs tests agains different Kafka versions and, therefore, test the compatibility.

Other than that, I think we can merge this PR.

Yeah I'll explore the failure today. Thanks for your quick review!

probot-autolabeler bot added io java kafka labels Jan 9, 2021

boyuanzz force-pushed the kafka_remove branch 2 times, most recently from 2acbf8a to 1435e98 Compare January 12, 2021 22:05

boyuanzz changed the title ~~[WIP] initial commit~~ [BEAM-11325] ReadFromKafkaDoFn should stop reading when topic/partition is removed or marked as stopped Jan 12, 2021

aromanenko-dev self-requested a review January 15, 2021 17:08

aromanenko-dev approved these changes Jan 19, 2021

View reviewed changes

boyuanzz force-pushed the kafka_remove branch from 1435e98 to 2afe9a4 Compare January 19, 2021 21:32

[BEAM-11325] ReadFromKafkaDoFn should stop reading when topic/partiti…

e8eed0b

…on is removed or marked as stopped.

boyuanzz force-pushed the kafka_remove branch from 2afe9a4 to e8eed0b Compare January 20, 2021 01:13

boyuanzz merged commit d2e1f69 into apache:master Jan 21, 2021

aromanenko-dev mentioned this pull request Feb 2, 2021

[BEAM-11325] Kafka Dynamic Read #13750

Merged

4 tasks

boyuanzz deleted the kafka_remove branch March 11, 2021 18:33

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-11325] ReadFromKafkaDoFn should stop reading when topic/partition is removed or marked as stopped #13710

[BEAM-11325] ReadFromKafkaDoFn should stop reading when topic/partition is removed or marked as stopped #13710

boyuanzz commented Jan 9, 2021 •

edited

aromanenko-dev commented Jan 15, 2021

aromanenko-dev left a comment

aromanenko-dev Jan 19, 2021

boyuanzz Jan 19, 2021

boyuanzz Jan 20, 2021

aromanenko-dev commented Jan 19, 2021

boyuanzz commented Jan 19, 2021

boyuanzz commented Jan 19, 2021

boyuanzz commented Jan 19, 2021

boyuanzz commented Jan 19, 2021

aromanenko-dev commented Jan 20, 2021 •

edited

aromanenko-dev commented Jan 21, 2021

aromanenko-dev commented Jan 21, 2021 •

edited

boyuanzz commented Jan 21, 2021

[BEAM-11325] ReadFromKafkaDoFn should stop reading when topic/partition is removed or marked as stopped #13710

[BEAM-11325] ReadFromKafkaDoFn should stop reading when topic/partition is removed or marked as stopped #13710

Conversation

boyuanzz commented Jan 9, 2021 • edited

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

aromanenko-dev commented Jan 15, 2021

aromanenko-dev left a comment

Choose a reason for hiding this comment

aromanenko-dev Jan 19, 2021

Choose a reason for hiding this comment

boyuanzz Jan 19, 2021

Choose a reason for hiding this comment

boyuanzz Jan 20, 2021

Choose a reason for hiding this comment

aromanenko-dev commented Jan 19, 2021

boyuanzz commented Jan 19, 2021

boyuanzz commented Jan 19, 2021

boyuanzz commented Jan 19, 2021

boyuanzz commented Jan 19, 2021

aromanenko-dev commented Jan 20, 2021 • edited

aromanenko-dev commented Jan 21, 2021

aromanenko-dev commented Jan 21, 2021 • edited

boyuanzz commented Jan 21, 2021

boyuanzz commented Jan 9, 2021 •

edited

aromanenko-dev commented Jan 20, 2021 •

edited

aromanenko-dev commented Jan 21, 2021 •

edited