You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Naireen (via GitHub)" <gi...@apache.org> on 2024/08/28 00:13:23 UTC
[PR] Fix Kafka with Redistribute [beam]
Naireen opened a new pull request, #32344:
URL: https://github.com/apache/beam/pull/32344
Check Argument to check if redistribute can be enabled was incorrect
Two ways to enable commits,
1) explicitly via commitOffsetsInFinalize()
2) 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.
Before this fix, it just meant customers werent 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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2335169004
Reminder, please take a look at this pr: @kennknowles @ahmedabu98
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #32344:
URL: https://github.com/apache/beam/pull/32344#discussion_r1757334704
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -2654,6 +2659,13 @@ public PCollection<KafkaRecord<K, V>> expand(PCollection<KafkaSourceDescriptor>
if (getRedistributeNumKeys() == 0) {
LOG.warn("This will create a key per record, which is sub-optimal for most use cases.");
}
+ // is another check here needed for with commit offsets
+ if (isCommitOffsetEnabled() || configuredKafkaCommit()) {
Review Comment:
That is true. I've updated to add that check.
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #32344:
URL: https://github.com/apache/beam/pull/32344#discussion_r1755209125
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -1697,11 +1696,17 @@ public PCollection<KafkaRecord<K, V>> expand(PBegin input) {
}
if (kafkaRead.isRedistributed()) {
- // fail here instead.
- checkArgument(
- kafkaRead.isCommitOffsetsInFinalizeEnabled(),
- "commitOffsetsInFinalize() can't be enabled with isRedistributed");
+ if (kafkaRead.isCommitOffsetsInFinalizeEnabled()) {
+ LOG.warn(
+ "commitOffsetsInFinalize() will not capture all work processed if set with withRedistribute()");
+ }
+ 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()");
Review Comment:
Ack, removed it.
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -1697,11 +1696,17 @@ public PCollection<KafkaRecord<K, V>> expand(PBegin input) {
}
if (kafkaRead.isRedistributed()) {
- // fail here instead.
- checkArgument(
- kafkaRead.isCommitOffsetsInFinalizeEnabled(),
- "commitOffsetsInFinalize() can't be enabled with isRedistributed");
+ if (kafkaRead.isCommitOffsetsInFinalizeEnabled()) {
+ LOG.warn(
+ "commitOffsetsInFinalize() will not capture all work processed if set with withRedistribute()");
+ }
+ if (Boolean.TRUE.equals(
Review Comment:
Yeah, it occurs a few more times in this file as well.
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2344413606
Run Java_Kafka_IO_Direct PreCommit
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2313908473
Reviewers: @scwhittle
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2316002328
> 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:
PCollection<KafkaSourceDescriptor> --> 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?)
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2314577533
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?
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #32344:
URL: https://github.com/apache/beam/pull/32344
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #32344:
URL: https://github.com/apache/beam/pull/32344#discussion_r1756577998
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -2654,6 +2659,13 @@ public PCollection<KafkaRecord<K, V>> expand(PCollection<KafkaSourceDescriptor>
if (getRedistributeNumKeys() == 0) {
LOG.warn("This will create a key per record, which is sub-optimal for most use cases.");
}
+ // is another check here needed for with commit offsets
+ if (isCommitOffsetEnabled() || configuredKafkaCommit()) {
Review Comment:
kafka autocommit can be ahead of records processed because the autocommit occurs when the consumer reads the data but the read data may be dropped if the commit to dataflow backend fails. Dataflow won't lose data in this case since it will reread from the previous offset which it stores internally but the auto commit offset will be further ahead than that offset.
That is true in all cases but if the pipeline is drained at that point the commitOffsetEnabled would be eventually correct matching what was actually processed by the pipeline while the configuredKafkaCommit would remain incorrect.
Separately should this log only be if there is redistribute withDuplicates allowed? If not, the redistribute is just like a reshuffle which will have normal semantics.
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2313913280
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).
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2317807893
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.
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2322775686
Perfect, thats great!
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #32344:
URL: https://github.com/apache/beam/pull/32344#discussion_r1762011686
##########
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() {
Review Comment:
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.
##########
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()");
Review Comment:
Done.
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2335169581
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`
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #32344:
URL: https://github.com/apache/beam/pull/32344#discussion_r1750170856
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -1697,11 +1696,17 @@ public PCollection<KafkaRecord<K, V>> expand(PBegin input) {
}
if (kafkaRead.isRedistributed()) {
- // fail here instead.
- checkArgument(
- kafkaRead.isCommitOffsetsInFinalizeEnabled(),
- "commitOffsetsInFinalize() can't be enabled with isRedistributed");
+ if (kafkaRead.isCommitOffsetsInFinalizeEnabled()) {
+ LOG.warn(
+ "commitOffsetsInFinalize() will not capture all work processed if set with withRedistribute()");
+ }
+ 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()");
Review Comment:
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.
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -1697,11 +1696,17 @@ public PCollection<KafkaRecord<K, V>> expand(PBegin input) {
}
if (kafkaRead.isRedistributed()) {
- // fail here instead.
- checkArgument(
- kafkaRead.isCommitOffsetsInFinalizeEnabled(),
- "commitOffsetsInFinalize() can't be enabled with isRedistributed");
+ if (kafkaRead.isCommitOffsetsInFinalizeEnabled()) {
+ LOG.warn(
+ "commitOffsetsInFinalize() will not capture all work processed if set with withRedistribute()");
Review Comment:
How about a little more detailed
Offsets committed due to usage to commitOffsetsInFinalize() may not capture all work processed due to use of withRedistribute()
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -1697,11 +1696,17 @@ public PCollection<KafkaRecord<K, V>> expand(PBegin input) {
}
if (kafkaRead.isRedistributed()) {
- // fail here instead.
- checkArgument(
- kafkaRead.isCommitOffsetsInFinalizeEnabled(),
- "commitOffsetsInFinalize() can't be enabled with isRedistributed");
+ if (kafkaRead.isCommitOffsetsInFinalizeEnabled()) {
+ LOG.warn(
+ "commitOffsetsInFinalize() will not capture all work processed if set with withRedistribute()");
+ }
+ if (Boolean.TRUE.equals(
Review Comment:
haven't seen this before, is this to handle null as a one-liner?
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -2654,6 +2659,13 @@ public PCollection<KafkaRecord<K, V>> expand(PCollection<KafkaSourceDescriptor>
if (getRedistributeNumKeys() == 0) {
LOG.warn("This will create a key per record, which is sub-optimal for most use cases.");
}
+ // is another check here needed for with commit offsets
+ if (isCommitOffsetEnabled() || configuredKafkaCommit()) {
Review Comment:
for configuredkafkacommit regardless of rediistribute the checkpoint will be ahead of processed records.
for isCommitOffsetEnabled the offset may be ahead or behind of the processing (since it's a separate pipeline branch) but it will be consistent on drain.
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #32344:
URL: https://github.com/apache/beam/pull/32344#issuecomment-2335169266
R: @scwhittle
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #32344:
URL: https://github.com/apache/beam/pull/32344#discussion_r1755209437
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -2654,6 +2659,13 @@ public PCollection<KafkaRecord<K, V>> expand(PCollection<KafkaSourceDescriptor>
if (getRedistributeNumKeys() == 0) {
LOG.warn("This will create a key per record, which is sub-optimal for most use cases.");
}
+ // is another check here needed for with commit offsets
+ if (isCommitOffsetEnabled() || configuredKafkaCommit()) {
Review Comment:
Your explanation for isCommitOffsetEnabled makes sense,
For configuredKafkaCommit, wouldn't the checkpoint always be behind processed records? We read messages, and then auto commit (default is every 5 seconds), so in between those 5 seconds, processing moves ahead of the last committed checkpoint?
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -1697,11 +1696,17 @@ public PCollection<KafkaRecord<K, V>> expand(PBegin input) {
}
if (kafkaRead.isRedistributed()) {
- // fail here instead.
- checkArgument(
- kafkaRead.isCommitOffsetsInFinalizeEnabled(),
- "commitOffsetsInFinalize() can't be enabled with isRedistributed");
+ if (kafkaRead.isCommitOffsetsInFinalizeEnabled()) {
+ LOG.warn(
+ "commitOffsetsInFinalize() will not capture all work processed if set with withRedistribute()");
Review Comment:
Done.
--
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: [PR] Fix Kafka with Redistribute [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #32344:
URL: https://github.com/apache/beam/pull/32344#discussion_r1758395798
##########
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()");
Review Comment:
update the log to reflect allow duplicates.
##########
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() {
Review Comment:
seems like it should be NoWarningsWith*No*AllowDuplicates
--
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