Fix Kafka with Redistribute #32344

Naireen · 2024-08-28T00:13:20Z

Change KafkaIO read withDuplicates to warn but still commit offsets if configured

Two ways to enable commits,

explicitly via commitOffsetsInFinalize()
Via consumer config ENABLE_AUTO_COMMIT_CONFIG=true

If the first is true, redistribute can't be enabled
If the second is true, the pipeline can still be passed, but isn't the most optimal, since if the runner wants to enable duplicates, there is no point of introducing the additional overhead of checkpointing messages within Kafka itself., though it is not semantically incorrect. The first option uses internal beam state to track which messages have been processed for preventing duplicates, which doesn't make sense if we also have a runner hint that duplicates can be allowed.

Before this fix, it just meant customers weren't able to set withRedistribute on the transform, even if commits weren't configured.

Updated tests to catch this.

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

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
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.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

Naireen · 2024-08-28T01:28:18Z

Reviewers: @scwhittle

github-actions · 2024-08-28T01:32:59Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.
R: @ahmedabu98 for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

scwhittle · 2024-08-28T07:48:47Z

One other benefit of the committed offsets is if it aids customer visibility in to progress of partitions as that can be queried/displayed external to dataflow.

Do we need to disable the committing of offsets? I can see the argument that it might not make sense from an exactly-once perspective but given that there are other reasons and that the customer is configuring it explicitly can we just perhaps log a warning that the offsets may not reflect all processed data but still perform them?

Naireen · 2024-08-28T18:27:51Z

One other benefit of the committed offsets is if it aids customer visibility in to progress of partitions as that can be queried/displayed external to dataflow.

That is a good point, so is an argument to still allow for it. What I dont understand though, is how does reshuffling/redistributing work with commiting offsets? Here is the current graph wiht offsets enabled (pulled from KafkaIO.java incase the formatting of the graph below isn't clear):

PCollection --> ParDo(ReadFromKafkaDoFn<KafkaSourceDescriptor, KV<KafkaSourceDescriptor, KafkaRecord>>) --> Reshuffle() --> Map(output KafkaRecord)
|
--> KafkaCommitOffset

At that point, if redistribute is enabled, does it make more sense to substitute the Reshuffle here for the Redistribute transform? Instead of inserting a reshuffle after the Map? (this would introduce another shuffle based on runner implementation)

If we go with the former approach, commits will still occur, but the commits of the commits can have duplicates (need to investigate what that can cause, or will it just be a no op if we attempt to commit the same offset internally in BEAM twice?)

scwhittle · 2024-08-29T14:15:19Z

My PR to remove the Reshuffle for the commit offsets was just merged. So I think the question on if it should be a redistribute if configured might be moot now. But we could disallow commit offsets and redistribute in the expand259 path since it is still an issue there.

Naireen · 2024-08-31T05:10:11Z

Perfect, thats great!

github-actions · 2024-09-07T12:13:53Z

Reminder, please take a look at this pr: @kennknowles @ahmedabu98

ahmedabu98 · 2024-09-07T12:15:05Z

R: @scwhittle

github-actions · 2024-09-07T12:16:13Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

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

scwhittle · 2024-09-09T12:28:21Z

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

+            LOG.warn(
+                "commitOffsetsInFinalize() will not capture all work processed if set with withRedistribute()");
+          }
+          if (Boolean.TRUE.equals(


haven't seen this before, is this to handle null as a one-liner?

Yeah, it occurs a few more times in this file as well.

scwhittle · 2024-09-09T12:30:33Z

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

+          if (Boolean.TRUE.equals(
+              kafkaRead.getConsumerConfig().get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG))) {
+            LOG.warn(
+                "config.ENABLE_AUTO_COMMIT_CONFIG doesn't need to be set with withRedistribute()");


It doesn't need to but it seems like it could still be desirable if they are using the offsets for monitoring.

Should we just remove this log? It seems the correctness is the same as if we were shuffling persistently though the window is perhaps larger if a drain is being performed.

Ack, removed it.

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

Naireen · 2024-09-11T18:41:06Z

Run Java_Kafka_IO_Direct PreCommit

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

scwhittle

Can you change the PR description to be a little clearer too, maybe
"Change KafkaIO read withDuplicates to warn but still commit offsets if configured"

scwhittle · 2024-09-13T08:09:34Z

sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java

@@ -648,6 +669,29 @@ public void testCommitOffsetsInFinalizeAndRedistributeWarnings() {
        "Offsets committed due to usage of commitOffsetsInFinalize() may not capture all work processed due to use of withRedistribute()");
  }

+  @Test
+  public void testCommitOffsetsInFinalizeAndRedistributeNoWarningsWithAllowDuplicates() {


seems like it should be NoWarningsWithNoAllowDuplicates

I do want to mention commits are enabled, since if we do not commit offsets, enabling allow duplicates also has no warning.
updated test name to NoWarningsWithNoAllowDuplicatesAndCommitOffsets, and updated the other test name so its more clear that the two are testing behaviour with and without allowDuplicates=true.

scwhittle · 2024-09-13T08:10:14Z

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

@@ -1696,7 +1696,7 @@ public PCollection<KafkaRecord<K, V>> expand(PBegin input) {
        }

        if (kafkaRead.isRedistributed()) {
-          if (kafkaRead.isCommitOffsetsInFinalizeEnabled()) {
+          if (kafkaRead.isCommitOffsetsInFinalizeEnabled() && kafkaRead.isAllowDuplicates()) {
            LOG.warn(
                "Offsets committed due to usage of commitOffsetsInFinalize() may not capture all work processed due to use of withRedistribute()");


update the log to reflect allow duplicates.

github-actions bot added java io kafka labels Aug 28, 2024

Naireen marked this pull request as ready for review August 28, 2024 01:25

github-actions bot added the Next Action: Reviewers label Aug 28, 2024

wasn't possible to enable this before

e016d25

Naireen force-pushed the fix_redistribute_with_kafka branch from e3af50a to aeb694c Compare August 30, 2024 21:55

adress comments

9c37865

Naireen force-pushed the fix_redistribute_with_kafka branch from aeb694c to 9c37865 Compare August 31, 2024 03:58

github-actions bot added the slow-review label Sep 7, 2024

scwhittle requested changes Sep 9, 2024

View reviewed changes

address review comments

3f6fc4e

Naireen force-pushed the fix_redistribute_with_kafka branch from c8f6dad to 3f6fc4e Compare September 11, 2024 20:18

scwhittle requested changes Sep 12, 2024

View reviewed changes

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java Outdated Show resolved Hide resolved

address comments

06e9c5a

scwhittle requested changes Sep 13, 2024

View reviewed changes

update test name

2bb6e3c

Naireen force-pushed the fix_redistribute_with_kafka branch from 87fe8e2 to 2bb6e3c Compare September 16, 2024 22:06

scwhittle approved these changes Sep 17, 2024

View reviewed changes

scwhittle merged commit 1b2d21a into apache:master Sep 17, 2024
20 checks passed

reeba212 pushed a commit to reeba212/beam that referenced this pull request Dec 4, 2024

Fix Kafka with Redistribute and commits enabled (apache#32344)

ace6aee

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix Kafka with Redistribute #32344

Fix Kafka with Redistribute #32344

Naireen commented Aug 28, 2024 •

edited

Loading

Naireen commented Aug 28, 2024

github-actions bot commented Aug 28, 2024

scwhittle commented Aug 28, 2024

Naireen commented Aug 28, 2024 •

edited

Loading

scwhittle commented Aug 29, 2024

Naireen commented Aug 31, 2024

github-actions bot commented Sep 7, 2024

ahmedabu98 commented Sep 7, 2024

github-actions bot commented Sep 7, 2024

scwhittle Sep 9, 2024

Naireen Sep 11, 2024

scwhittle Sep 9, 2024

Naireen Sep 11, 2024

Naireen commented Sep 11, 2024

scwhittle left a comment

scwhittle Sep 13, 2024

Naireen Sep 16, 2024

scwhittle Sep 13, 2024

Naireen Sep 16, 2024

Fix Kafka with Redistribute #32344

Fix Kafka with Redistribute #32344

Conversation

Naireen commented Aug 28, 2024 • edited Loading

GitHub Actions Tests Status (on master branch)

Naireen commented Aug 28, 2024

github-actions bot commented Aug 28, 2024

scwhittle commented Aug 28, 2024

Naireen commented Aug 28, 2024 • edited Loading

scwhittle commented Aug 29, 2024

Naireen commented Aug 31, 2024

github-actions bot commented Sep 7, 2024

ahmedabu98 commented Sep 7, 2024

github-actions bot commented Sep 7, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Naireen commented Sep 11, 2024

scwhittle left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Naireen commented Aug 28, 2024 •

edited

Loading

Naireen commented Aug 28, 2024 •

edited

Loading