You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "tomstepp (via GitHub)" <gi...@apache.org> on 2025/01/14 19:24:04 UTC
[PR] Add offset-based deduplication to Unbounded Source and Dataflow runner. [beam]
tomstepp opened a new pull request, #33591:
URL: https://github.com/apache/beam/pull/33591
Add offset-based deduplication to Unbounded Source and Dataflow runner.
------------------------
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #33591:
URL: https://github.com/apache/beam/pull/33591#issuecomment-2612078065
failures appear unrelated
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1925656131
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -203,6 +224,11 @@ public byte[] getCurrentRecordId() throws NoSuchElementException {
return EMPTY;
}
+ /* Returns the offset for the current record of this unbounded reader. */
+ public byte[] getCurrentRecordOffset() {
+ return EMPTY;
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1924239234
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillSink.java:
##########
@@ -215,7 +215,13 @@ public long add(WindowedValue<T> data) throws IOException {
.setMetadata(metadata);
keyedOutput.addMessages(builder.build());
keyedOutput.addMessagesIds(id);
- return (long) key.size() + value.size() + metadata.size() + id.size();
+
+ ByteString offset = ByteString.copyFrom(context.getCurrentRecordOffset());
Review Comment:
would be nice to avoid allocation if no current record since that is common
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -139,6 +146,11 @@ public void finalizeCheckpoint() throws IOException {
// nothing to do
}
}
+
+ /* Get offset limit for unbounded source split checkpoint. */
+ default byte[] getOffsetLimit() throws IOException {
Review Comment:
maybe we should just throw a runtime exception? it seems like this shouldn't have to do IO and IOException needs to be handled at callers if method is annoated as throwing non-runtime exception
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -93,6 +93,13 @@ public boolean requiresDeduping() {
return false;
}
+ /**
+ * Returns whether this source is configured for offset-based deduplication by the runner.
+ */
+ public boolean offsetDeduplication() {
Review Comment:
can this be named as a property of the source instead? The difference being that it is available to runners to use for optimization if desired but it isn't required that the runner does.
so maybe this should be named isOffsetBased with a comment that if it returns true then the source needs to provide offsets that are unique for each element, and can be ordered lexicographically for both the UnboundedReaders it creates and the checkpoint marks it vends must have an offset greater orr equal to all elements read and less than the next element (or whatever the edge-case semantics are).
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1919418801
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -194,6 +194,13 @@ public boolean workIsFailed() {
return Optional.ofNullable(work).map(Work::isFailed).orElse(false);
}
+ public byte[] getCurrentRecordOffset() {
+ if (activeReader == null) {
+ return new byte[0];
Review Comment:
I used the static singleton for now. Perhaps we can continue with it based on notes below.
I previously tried annotating as nullable but received errors that it is not compatible with primitive types like byte. Without nullable annotation, returning null is also marked as an error. I have not done a lot of Beam or Java coding, so I possibly misunderstood something.
However, I think empty bytes can/should be considered invalid, so it's okay to use empty as invalid rather than both empty and null. Wdyt?
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1925655287
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -194,6 +195,13 @@ public boolean workIsFailed() {
return Optional.ofNullable(work).map(Work::isFailed).orElse(false);
}
+ public byte[] getCurrentRecordOffset() {
+ if (activeReader == null) {
+ return NO_RECORD_OFFSET;
+ }
+ return activeReader.getCurrentRecordOffset();
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1919418962
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -139,6 +146,11 @@ public void finalizeCheckpoint() throws IOException {
// nothing to do
}
}
+
+ /* Get offset limit for unbounded source split checkpoint. */
+ default byte[] getOffsetLimit() {
+ return new byte[0];
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] Add offset-based deduplication to Unbounded Source and Dataflow runner. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33591:
URL: https://github.com/apache/beam/pull/33591#issuecomment-2590937313
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] Add offset-based deduplication to Unbounded Source and Dataflow runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on PR #33591:
URL: https://github.com/apache/beam/pull/33591#issuecomment-2590934633
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1926729156
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillSink.java:
##########
@@ -214,8 +214,27 @@ public long add(WindowedValue<T> data) throws IOException {
.setData(value)
.setMetadata(metadata);
keyedOutput.addMessages(builder.build());
+
+ long offsetSize = 0;
+ if (context.offsetBasedDeduplicationEnabled()) {
+ byte[] rawId = context.getCurrentRecordId();
+ if (rawId.length == 0) {
+ throw new RuntimeException("Unexpected empty record ID while offset-based deduplication enabled.");
+ }
+ id = ByteString.copyFrom(rawId);
Review Comment:
should we prefer this id or the id from ValueWithRecordId if both are set?
or we could make it not valid to have both set or expect them to be the same if both are set.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -93,6 +93,21 @@ public boolean requiresDeduping() {
return false;
}
+ /**
+ * If offsetBasedDeduplicationEnabled returns true, then the UnboundedSource needs to provide the
+ * following:
+ *
+ * <ul>
+ * <li>UnboundedReader which provides offsets that are unique for each element and
+ * lexicographically ordered.
+ * <li>CheckpointMark which provides an offset greater than all elements read and less than or
+ * equal to the next offset that will be read.
+ * </ul>
+ */
+ public boolean offsetBasedDeduplicationEnabled() {
Review Comment:
sorry to still nit on this, but Enabled could be confusing if the runner doesn't support it or use the offsets.
How about offsetBasedDeduplicationSupported? update the execution context method as well
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -139,6 +146,11 @@ public void finalizeCheckpoint() throws IOException {
// nothing to do
}
}
+
+ /* Get offset limit for unbounded source split checkpoint. */
+ default byte[] getOffsetLimit() {
Review Comment:
Yup that's what I was thinking, thanks
--
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] Add offset-based deduplication to Unbounded Source and Dataflow runner. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1918689052
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -194,6 +194,13 @@ public boolean workIsFailed() {
return Optional.ofNullable(work).map(Work::isFailed).orElse(false);
}
+ public byte[] getCurrentRecordOffset() {
+ if (activeReader == null) {
+ return new byte[0];
Review Comment:
have an static singleton of this?
NO_RECORD_OFFSET
or perhaps better to make this Nullable and return null? If caller expects a reader to be present this would allow for them to check that instead of possibly using the bad offset.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -139,6 +146,11 @@ public void finalizeCheckpoint() throws IOException {
// nothing to do
}
}
+
+ /* Get offset limit for unbounded source split checkpoint. */
+ default byte[] getOffsetLimit() {
+ return new byte[0];
Review Comment:
return EMPTY
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -139,6 +146,11 @@ public void finalizeCheckpoint() throws IOException {
// nothing to do
}
}
+
+ /* Get offset limit for unbounded source split checkpoint. */
+ default byte[] getOffsetLimit() {
Review Comment:
should we require this to be overridden (ie byte[0] is not valid) if offsetDeduplication is true?
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1927419071
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -93,6 +93,21 @@ public boolean requiresDeduping() {
return false;
}
+ /**
+ * If offsetBasedDeduplicationEnabled returns true, then the UnboundedSource needs to provide the
+ * following:
+ *
+ * <ul>
+ * <li>UnboundedReader which provides offsets that are unique for each element and
+ * lexicographically ordered.
+ * <li>CheckpointMark which provides an offset greater than all elements read and less than or
+ * equal to the next offset that will be read.
+ * </ul>
+ */
+ public boolean offsetBasedDeduplicationEnabled() {
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #33591:
URL: https://github.com/apache/beam/pull/33591
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1924343981
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillSink.java:
##########
@@ -215,7 +215,13 @@ public long add(WindowedValue<T> data) throws IOException {
.setMetadata(metadata);
keyedOutput.addMessages(builder.build());
keyedOutput.addMessagesIds(id);
- return (long) key.size() + value.size() + metadata.size() + id.size();
+
+ ByteString offset = ByteString.copyFrom(context.getCurrentRecordOffset());
Review Comment:
Sounds good, updated to only allocate if non-zero offset.
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1925902349
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -194,6 +195,13 @@ public boolean workIsFailed() {
return Optional.ofNullable(work).map(Work::isFailed).orElse(false);
}
+ public byte[] getCurrentRecordOffset() {
+ if (activeReader == null) {
+ return NO_RECORD_OFFSET;
+ }
+ return activeReader.getCurrentRecordOffset();
Review Comment:
Also updated to throw exception on these methods if offset dedup is off. This removed the prior usage of returning empties, but please lmk if you think the approach should be adjusted.
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1925655873
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -93,6 +93,21 @@ public boolean requiresDeduping() {
return false;
}
+ /**
+ * If isOffsetDeduplication returns true, then the UnboundedSource needs to
+ * provide the following:
+ *
+ * <ul>
+ * <li>UnboundedReader which provides offsets that are unique for each
+ * element and lexicographically ordered.</li>
+ * <li>CheckpointMark which provides an offset greater than all elements
+ * read and less than or equal to the next offset that will be read.</li>
+ * </ul>
+ */
+ public boolean isOffsetDeduplication() {
Review Comment:
Updated to offsetBasedDeduplicationEnabled
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1925031478
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -194,6 +195,13 @@ public boolean workIsFailed() {
return Optional.ofNullable(work).map(Work::isFailed).orElse(false);
}
+ public byte[] getCurrentRecordOffset() {
+ if (activeReader == null) {
+ return NO_RECORD_OFFSET;
+ }
+ return activeReader.getCurrentRecordOffset();
Review Comment:
maybe we should do
if (activeReader == null || !activeReader.getSource().providesRecordOffsets()) {
return NO_RECORD_OFFSET;
}
return activeReader.getCurrentRecordOffset();
Then we could change the reader implementation to verify the default reader impl to just throw an exception that it needs to be overridden if providesRecordOffsets is set to true.
I'm just trying to make it not possible to set providesRecordOffsets but forget to implement the necessary methods.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -93,6 +93,21 @@ public boolean requiresDeduping() {
return false;
}
+ /**
+ * If isOffsetDeduplication returns true, then the UnboundedSource needs to
+ * provide the following:
+ *
+ * <ul>
+ * <li>UnboundedReader which provides offsets that are unique for each
+ * element and lexicographically ordered.</li>
+ * <li>CheckpointMark which provides an offset greater than all elements
+ * read and less than or equal to the next offset that will be read.</li>
+ * </ul>
+ */
+ public boolean isOffsetDeduplication() {
Review Comment:
maybe providesRecordOffsets? supportsOffsetBasedDeduping?
trying to think of something that is easy to read when in code
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -203,6 +224,11 @@ public byte[] getCurrentRecordId() throws NoSuchElementException {
return EMPTY;
}
+ /* Returns the offset for the current record of this unbounded reader. */
+ public byte[] getCurrentRecordOffset() {
+ return EMPTY;
Review Comment:
see other comment, seems we could throw like the default checkpoint method if we guard at caller
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1919418801
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -194,6 +194,13 @@ public boolean workIsFailed() {
return Optional.ofNullable(work).map(Work::isFailed).orElse(false);
}
+ public byte[] getCurrentRecordOffset() {
+ if (activeReader == null) {
+ return new byte[0];
Review Comment:
I used the static singleton for now. Perhaps we can continue with it based on notes below.
I previously tried annotating as nullable but received errors that it is not compatible with primitive types like byte. Without nullable annotation, returning null is also marked as an error. I have not done a lot of Beam Java coding, so I possibly misunderstood something.
However, I think empty bytes can/should be considered invalid, so it's okay to use empty as invalid rather than both empty and null. Wdyt?
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1919431625
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -139,6 +146,11 @@ public void finalizeCheckpoint() throws IOException {
// nothing to do
}
}
+
+ /* Get offset limit for unbounded source split checkpoint. */
+ default byte[] getOffsetLimit() {
Review Comment:
That sounds good. IIUC this means the default implementation would throw an exception and the runner (StreamingModeExecutionContext for Dataflow) must only call this method if the source enables offset deduplication. Updated PR to match this suggestion.
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1919418801
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -194,6 +194,13 @@ public boolean workIsFailed() {
return Optional.ofNullable(work).map(Work::isFailed).orElse(false);
}
+ public byte[] getCurrentRecordOffset() {
+ if (activeReader == null) {
+ return new byte[0];
Review Comment:
I used the static singleton.
I previously tried annotating as nullable but received errors that it is not compatible with primitive types like byte. Without nullable annotation, returning null is also marked as an error. I have not done a lot of Beam Java coding, so I possibly misunderstood something.
However, I think empty bytes can/should be considered invalid, so it's okay to use empty as invalid rather than both empty and null. Wdyt?
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1924345498
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -139,6 +146,11 @@ public void finalizeCheckpoint() throws IOException {
// nothing to do
}
}
+
+ /* Get offset limit for unbounded source split checkpoint. */
+ default byte[] getOffsetLimit() throws IOException {
Review Comment:
Ah I see the difference now, updated to RuntimeException.
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1924344487
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:
##########
@@ -93,6 +93,13 @@ public boolean requiresDeduping() {
return false;
}
+ /**
+ * Returns whether this source is configured for offset-based deduplication by the runner.
+ */
+ public boolean offsetDeduplication() {
Review Comment:
Updated to isOffsetDeduplication with more detailed comment.
--
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] Add offset-based deduplication to Unbounded Source and Dataflow Runner. [beam]
Posted by "tomstepp (via GitHub)" <gi...@apache.org>.
tomstepp commented on code in PR #33591:
URL: https://github.com/apache/beam/pull/33591#discussion_r1927532919
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillSink.java:
##########
@@ -214,8 +214,27 @@ public long add(WindowedValue<T> data) throws IOException {
.setData(value)
.setMetadata(metadata);
keyedOutput.addMessages(builder.build());
+
+ long offsetSize = 0;
+ if (context.offsetBasedDeduplicationEnabled()) {
+ byte[] rawId = context.getCurrentRecordId();
+ if (rawId.length == 0) {
+ throw new RuntimeException("Unexpected empty record ID while offset-based deduplication enabled.");
+ }
+ id = ByteString.copyFrom(rawId);
Review Comment:
Updated to not expect ID from ValueWithRecordId.
--
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