github-actions[bot] commented on code in PR #41328:
URL: https://github.com/apache/beam/pull/41328#discussion_r4572247144
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/PubsubSink.java:
##########
@@ -172,6 +173,15 @@ public long add(WindowedValue<T> data) throws IOException {
if (formatted.getAttributeMap() != null) {
pubsubMessageBuilder.putAllAttributes(formatted.getAttributeMap());
}
+ if (idLabel != null
+ && !idLabel.isEmpty()
+ &&
!pubsubMessageBuilder.getAttributesMap().containsKey(idLabel)) {
+ // Generate a unique id that is stable across retries of the same
element, so that
+ // redeliveries of a bundle can be deduplicated by Pub/Sub.
+ pubsubMessageBuilder.putAttributes(
+ idLabel,
+
UUID.nameUUIDFromBytes(pubsubMessageBuilder.build().toByteArray()).toString());
Review Comment:
**🔴 Critical**: There are two major issues with setting `idLabel` here:
1. **Wrong component for #18023:**
`org.apache.beam.runners.dataflow.worker.PubsubSink` is Dataflow's
**streaming** Windmill sink (used with `StreamingModeExecutionContext`), where
`idLabel` is already forwarded on the bundle via
`Windmill.PubSubMessageBundle.Builder.setIdLabel(idLabel)` (line 153) for
Windmill to populate. Issue #18023 (`BEAM-361`) is about the **batch
(bounded)** Pub/Sub sink in
`sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java`
(`PubsubIO.Write.PubsubBoundedWriter`), where `startBundle` currently ignores
`getIdAttribute()` (`// NOTE: idAttribute is ignored.`) and `processElement`
always sets `recordId` to `null`.
2. **Data loss for legitimate duplicate payloads:**
`UUID.nameUUIDFromBytes(pubsubMessageBuilder.build().toByteArray())` hashes
only the message payload and attributes. Any two distinct elements with the
same payload and attributes (even at different timestamps or in different
bundles) will receive the exact same `idLabel` UUID and will be silently
dropped as duplicates by downstream Pub/Sub readers deduplicating on
`idAttribute`. (For comparison, see how `PubsubUnboundedSink` assigns
`UUID.randomUUID().toString()` per element before a `GroupByKey`/checkpoint
boundary so retries reuse the same ID without content collisions.)
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/PubsubSinkTest.java:
##########
@@ -131,6 +137,115 @@ public void testEmptyParseFn() throws Exception {
testWriteWith("");
}
+ private static class FormatFn extends SimpleFunction<String, PubsubMessage> {
+ private final Map<String, String> attributes;
+
+ FormatFn(Map<String, String> attributes) {
+ this.attributes = attributes;
+ }
+
+ @Override
+ public PubsubMessage apply(String input) {
+ return new PubsubMessage(input.getBytes(StandardCharsets.UTF_8),
attributes);
+ }
+ }
+
+ private Windmill.WorkItemCommitRequest.Builder writeWithFormatFn(
+ SimpleFunction<String, PubsubMessage> formatFn) throws Exception {
+ Windmill.WorkItemCommitRequest.Builder outputBuilder =
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(ByteString.copyFromUtf8("key"))
+ .setWorkToken(0);
+ when(mockContext.getKeyOutputBuilder()).thenReturn(outputBuilder);
+
+ PubsubSink<String> sink =
+ new PubsubSink<>(
+ "topic",
+ "ts",
+ "id",
+ WindowedValues.getFullCoder(StringUtf8Coder.of(),
IntervalWindow.getCoder()),
+ formatFn,
+ true,
+ mockContext);
+
+ Sink.SinkWriter<WindowedValue<String>> writer = sink.writer();
+ writer.add(WindowedValues.timestampedValueInGlobalWindow("e0", new
Instant(0)));
+ writer.add(WindowedValues.timestampedValueInGlobalWindow("e0", new
Instant(1000)));
Review Comment:
**🟠High**: This test illustrates the collision problem: two distinct
elements (`"e0"` at `Instant(0)` and `"e0"` at `Instant(1000)`) are written to
the sink, and `testWriteSetsIdLabelWhenMissing` asserts that both messages are
emitted with the exact same `expectedId` in their `"id"` attribute. A
downstream consumer reading with `withIdAttribute("id")` will treat the second
message at `Instant(1000)` as a duplicate of the first and drop it.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
