You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "sjvanrossum (via GitHub)" <gi...@apache.org> on 2025/11/28 12:40:36 UTC
[PR] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
sjvanrossum opened a new pull request, #36935:
URL: https://github.com/apache/beam/pull/36935
Replace the restriction trackers in `ReadFromKafkaDoFn` with `UnsplittableRestrictionTracker<OffsetRange, Long>` to prevent non-checkpointing splits. Consumer groups must not consume from the same `TopicPartition` simultaneously and this design detail was used to limit the creation of `Consumer` instances in `ReadFromKafkaDoFn`, but failed to be upheld by its split results.
------------------------
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #36935:
URL: https://github.com/apache/beam/pull/36935
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on code in PR #36935:
URL: https://github.com/apache/beam/pull/36935#discussion_r2572076012
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/UnsplittableRestrictionTracker.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+/**
+ * A {@link RestrictionTracker} for wrapping a {@link RestrictionTracker} with unsplittable
+ * restrictions.
+ *
+ * <p>A restriction is considered unsplittable when restrictions of an element must not be processed
+ * simultaneously (e.g., Kafka topic partition).
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class UnsplittableRestrictionTracker<RestrictionT, PositionT>
+ extends RestrictionTracker<RestrictionT, PositionT> implements RestrictionTracker.HasProgress {
+ private final RestrictionTracker<RestrictionT, PositionT> tracker;
+
+ public UnsplittableRestrictionTracker(RestrictionTracker<RestrictionT, PositionT> tracker) {
+ this.tracker = tracker;
+ }
+
+ @Override
+ public boolean tryClaim(PositionT position) {
+ return tracker.tryClaim(position);
+ }
+
+ @Override
+ public RestrictionT currentRestriction() {
+ return tracker.currentRestriction();
+ }
+
+ @Override
+ public SplitResult<RestrictionT> trySplit(double fractionOfRemainder) {
+ return fractionOfRemainder < 1.0 ? null : tracker.trySplit(fractionOfRemainder);
Review Comment:
Fixed.
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #36935:
URL: https://github.com/apache/beam/pull/36935#discussion_r2571677163
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java:
##########
@@ -108,6 +109,12 @@
*
* <h4>Splitting</h4>
*
+ * <p>Consumer groups must not consume from the same {@link TopicPartition} simultaneously. Doing so
+ * may arbitrarily overwrite a consumer group's committed offset for a {@link TopicPartition}.
Review Comment:
So I think that this could work, but would we rather support processing distinct portions of the same partition by caching more than 1 consumer? I think we'd have to change the code to more explicitly acquire and release a consumer.
Though batch isn't particularly common given we haven't hit this before, this could maybe be useful even for streaming for processing large backlogs if we made the initial splits per partition something like [s, (s+f)/2) [(s+f)/2, f) [f, streaming tail] where s is start position, f is current position.
This is assuming that consumers are independent and that this won't screw up caches or something on the server and kill performance.
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on PR #36935:
URL: https://github.com/apache/beam/pull/36935#issuecomment-3666781384
@scwhittle I validated the changes in this PR using Dataflow Runner v2 with a pipeline that closely resembles the integration test in this PR. A build of the pipeline based on 2.69.0 fails with a `ConcurrentModificationException` and a build of the pipeline based on 2.69.0 with these changes succeeds.
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on code in PR #36935:
URL: https://github.com/apache/beam/pull/36935#discussion_r2572134546
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java:
##########
@@ -108,6 +109,12 @@
*
* <h4>Splitting</h4>
*
+ * <p>Consumer groups must not consume from the same {@link TopicPartition} simultaneously. Doing so
+ * may arbitrarily overwrite a consumer group's committed offset for a {@link TopicPartition}.
Review Comment:
I'd prefer that over `UnsplittableRestrictionTracker`, but we'd still need to prevent splitting when `enable.auto.commit` is set. If runners split too aggressively during processing then there's the risk of overwhelming a broker with connections, but I'm not too concerned about that.
I've asked the release manager to weigh in on whether they want to cherry-pick this change into 2.70.0 since it would unblock a user. If not, I'll add those changes to this PR.
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on PR #36935:
URL: https://github.com/apache/beam/pull/36935#issuecomment-3666702129
Run Java_GCP_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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #36935:
URL: https://github.com/apache/beam/pull/36935#issuecomment-3589658281
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, 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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #36935:
URL: https://github.com/apache/beam/pull/36935#discussion_r2576562216
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java:
##########
@@ -108,6 +109,14 @@
*
* <h4>Splitting</h4>
*
+ * <p>Consumer group members must not consume from the same {@link TopicPartition} simultaneously
+ * when {@code enable.auto.commit} is set. Doing so may arbitrarily overwrite a consumer group's
+ * committed offset for a {@link TopicPartition}. Restriction trackers for a {@link
+ * KafkaSourceDescriptor} are wrapped as {@link UnsplittableRestrictionTracker<OffsetRange, Long>}
+ * and will only return a non-null {@link org.apache.beam.sdk.transforms.splittabledofn.SplitResult}
+ * for a checkpoint. This ensures consistent behavior when {@code enable.auto.commit} is set and
Review Comment:
nit: "ensures consistent" is too strong I think? There may still be parallel scheduling on different VMs during scaling etc.
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on PR #36935:
URL: https://github.com/apache/beam/pull/36935#issuecomment-3666700637
Run Java 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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #36935:
URL: https://github.com/apache/beam/pull/36935#issuecomment-3589194917
## Summary of Changes
Hello @sjvanrossum, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request addresses a critical issue in Apache Beam's KafkaIO connector where `ReadFromKafkaDoFn` could inadvertently allow non-checkpointing splits for Kafka topic partitions. Such splits could lead to multiple consumers processing the same partition simultaneously, resulting in incorrect committed offsets and potential data loss or inconsistencies. By introducing an `UnsplittableRestrictionTracker`, the PR ensures that Kafka topic partitions are processed exclusively, safeguarding the integrity of consumer group offsets and improving the reliability of Kafka data ingestion.
### Highlights
* **New Restriction Tracker**: Introduced `UnsplittableRestrictionTracker` to enforce that certain restrictions, like those for Kafka topic partitions, cannot be split during processing except for checkpointing.
* **KafkaIO Integration**: `ReadFromKafkaDoFn` now uses `UnsplittableRestrictionTracker` to wrap its internal `OffsetRangeTracker` and `GrowableOffsetRangeTracker`, preventing unintended concurrent consumption of Kafka topic partitions.
* **Javadoc Update**: Clarified documentation in `GrowableOffsetRangeTracker` and `ReadFromKafkaDoFn` regarding the handling of Kafka topic partition offsets and the purpose of unsplittable restrictions.
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
<b>You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the [Gemini Code Assist IDE Extension](https://cloud.google.com/products/gemini/code-assist).</b>
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #36935:
URL: https://github.com/apache/beam/pull/36935#discussion_r2571578547
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/UnsplittableRestrictionTracker.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+/**
+ * A {@link RestrictionTracker} for wrapping a {@link RestrictionTracker} with unsplittable
+ * restrictions.
+ *
+ * <p>A restriction is considered unsplittable when restrictions of an element must not be processed
+ * simultaneously (e.g., Kafka topic partition).
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class UnsplittableRestrictionTracker<RestrictionT, PositionT>
+ extends RestrictionTracker<RestrictionT, PositionT> implements RestrictionTracker.HasProgress {
+ private final RestrictionTracker<RestrictionT, PositionT> tracker;
+
+ public UnsplittableRestrictionTracker(RestrictionTracker<RestrictionT, PositionT> tracker) {
+ this.tracker = tracker;
+ }
+
+ @Override
+ public boolean tryClaim(PositionT position) {
+ return tracker.tryClaim(position);
+ }
+
+ @Override
+ public RestrictionT currentRestriction() {
+ return tracker.currentRestriction();
+ }
+
+ @Override
+ public SplitResult<RestrictionT> trySplit(double fractionOfRemainder) {
+ return fractionOfRemainder < 1.0 ? null : tracker.trySplit(fractionOfRemainder);
Review Comment:
it seems like 0 should also be handled specially, from base comment:
```
* @return a {@link SplitResult} if a split was possible, otherwise returns {@code null}. If the
* {@code fractionOfRemainder == 0}, a {@code null} result MUST imply that the restriction
* tracker is done and there is no more work left to do.
```
0 also seems ok to delegate since we shouldn't be doing parallel processing in that case either
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java:
##########
@@ -108,6 +109,12 @@
*
* <h4>Splitting</h4>
*
+ * <p>Consumer groups must not consume from the same {@link TopicPartition} simultaneously. Doing so
+ * may arbitrarily overwrite a consumer group's committed offset for a {@link TopicPartition}.
Review Comment:
this committed offset is mostly for external monitoring correct? Beam is maintaining it's own offsets for reading for correctness reasons and I think that it could handle reading the same partition in parallel if it wasn't sharing a consumer.
If the mode of committing the offset in bundle finalizations, the offset being committed could be out-of-order or stale anyway.
So I think mainly we are preventing the splitting because the current caching strategy doesn't work well with it. I'm also not sure if it would actually improve throughput to read from the same partition from different threads but if we are reading fixed non-overlapping ranges it seems like it possibly could.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/UnsplittableRestrictionTracker.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+/**
+ * A {@link RestrictionTracker} for wrapping a {@link RestrictionTracker} with unsplittable
+ * restrictions.
+ *
+ * <p>A restriction is considered unsplittable when restrictions of an element must not be processed
+ * simultaneously (e.g., Kafka topic partition).
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class UnsplittableRestrictionTracker<RestrictionT, PositionT>
+ extends RestrictionTracker<RestrictionT, PositionT> implements RestrictionTracker.HasProgress {
+ private final RestrictionTracker<RestrictionT, PositionT> tracker;
+
+ public UnsplittableRestrictionTracker(RestrictionTracker<RestrictionT, PositionT> tracker) {
+ this.tracker = tracker;
+ }
+
+ @Override
+ public boolean tryClaim(PositionT position) {
+ return tracker.tryClaim(position);
+ }
+
+ @Override
+ public RestrictionT currentRestriction() {
+ return tracker.currentRestriction();
+ }
+
+ @Override
+ public SplitResult<RestrictionT> trySplit(double fractionOfRemainder) {
Review Comment:
base method is @Nullable, if you mark this nullable here, can you remove the suppression above? If for some reason it still doesn't work can the suppression be limited to this function?
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #36935:
URL: https://github.com/apache/beam/pull/36935#issuecomment-3669345511
Test failure is unrelated flake
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on code in PR #36935:
URL: https://github.com/apache/beam/pull/36935#discussion_r2572075542
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/UnsplittableRestrictionTracker.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+/**
+ * A {@link RestrictionTracker} for wrapping a {@link RestrictionTracker} with unsplittable
+ * restrictions.
+ *
+ * <p>A restriction is considered unsplittable when restrictions of an element must not be processed
+ * simultaneously (e.g., Kafka topic partition).
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class UnsplittableRestrictionTracker<RestrictionT, PositionT>
+ extends RestrictionTracker<RestrictionT, PositionT> implements RestrictionTracker.HasProgress {
+ private final RestrictionTracker<RestrictionT, PositionT> tracker;
+
+ public UnsplittableRestrictionTracker(RestrictionTracker<RestrictionT, PositionT> tracker) {
+ this.tracker = tracker;
+ }
+
+ @Override
+ public boolean tryClaim(PositionT position) {
+ return tracker.tryClaim(position);
+ }
+
+ @Override
+ public RestrictionT currentRestriction() {
+ return tracker.currentRestriction();
+ }
+
+ @Override
+ public SplitResult<RestrictionT> trySplit(double fractionOfRemainder) {
Review Comment:
Fixed in `UnsplittableRestrictionTracker`. I had a look at doing the same for `OffsetRangeTracker` and `GrowableOffsetRangeTracker` and it looks like there's a synchronization issue between `tryClaim` and `getProgress`. `FnApiDoFnRunner` calls `tryClaim` when `splitLock` is not held (during `processElement`) and modifies `lastAttemptedOffset`. `FnApiDoFnRunner` also calls `getProgress` when `splitLock` is held so this prevents the observation of different values for `range`, but it does not prevent the observation of different values for `lastAttemptedOffset`. It looks benign since the `lastAttemptedOffset` is only read to perform a null check and then read again to use its value, but the null checker complains about it nonetheless. I'll patch that in a different PR.
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on code in PR #36935:
URL: https://github.com/apache/beam/pull/36935#discussion_r2576870890
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java:
##########
@@ -108,6 +109,14 @@
*
* <h4>Splitting</h4>
*
+ * <p>Consumer group members must not consume from the same {@link TopicPartition} simultaneously
+ * when {@code enable.auto.commit} is set. Doing so may arbitrarily overwrite a consumer group's
+ * committed offset for a {@link TopicPartition}. Restriction trackers for a {@link
+ * KafkaSourceDescriptor} are wrapped as {@link UnsplittableRestrictionTracker<OffsetRange, Long>}
+ * and will only return a non-null {@link org.apache.beam.sdk.transforms.splittabledofn.SplitResult}
+ * for a checkpoint. This ensures consistent behavior when {@code enable.auto.commit} is set and
Review Comment:
Fixed. The intent wasn't to comment on consistency guarantees, but the phrasing is off.
--
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] [KafkaIO] Make ReadFromKafkaDoFn restriction trackers unsplittable [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on PR #36935:
URL: https://github.com/apache/beam/pull/36935#issuecomment-3596378243
> Could we add some batch Kafka pipeline integration test to validate this (and catch other possible issues)?
Added (as you may have seen), but the integration test passes with and without these changes. I'm hoping to validate this patch on the Prism runner later today.
--
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