From f0c92c7a772397c3c6359b8478c6ed48c117cd30 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Wed, 15 Oct 2025 14:40:15 +0200 Subject: [PATCH] revert outputWindowedValue changes as there is outputBuilder --- ...oundedSplittableProcessElementInvoker.java | 35 +--- .../beam/runners/core/SimpleDoFnRunner.java | 130 ------------- .../SplittableParDoViaKeyedWorkItems.java | 21 --- .../org/apache/beam/sdk/transforms/DoFn.java | 31 ---- .../beam/sdk/transforms/DoFnTester.java | 64 ------- .../SplittableParDoNaiveBounded.java | 48 ----- .../beam/fn/harness/FnApiDoFnRunner.java | 171 +----------------- .../org/apache/beam/sdk/io/kafka/KafkaIO.java | 14 +- 8 files changed, 5 insertions(+), 509 deletions(-) diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java index 9bda4dd2cbca..767673959663 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java @@ -424,24 +424,6 @@ public void outputWindowedValue( outputReceiver.output(mainOutputTag, WindowedValues.of(value, timestamp, windows, paneInfo)); } - @Override - public void outputWindowedValue( - OutputT value, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - noteOutput(); - if (watermarkEstimator instanceof TimestampObservingWatermarkEstimator) { - ((TimestampObservingWatermarkEstimator) watermarkEstimator).observeTimestamp(timestamp); - } - outputReceiver.output( - mainOutputTag, - WindowedValues.of( - value, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } - @Override public void output(TupleTag tag, T value) { outputWithTimestamp(tag, value, element.getTimestamp()); @@ -460,26 +442,11 @@ public void outputWindowedValue( Instant timestamp, Collection windows, PaneInfo paneInfo) { - outputWindowedValue(tag, value, timestamp, windows, paneInfo, null, null); - } - - @Override - public void outputWindowedValue( - TupleTag tag, - T value, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { noteOutput(); if (watermarkEstimator instanceof TimestampObservingWatermarkEstimator) { ((TimestampObservingWatermarkEstimator) watermarkEstimator).observeTimestamp(timestamp); } - outputReceiver.output( - tag, - WindowedValues.of( - value, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); + outputReceiver.output(tag, WindowedValues.of(value, timestamp, windows, paneInfo)); } private void noteOutput() { diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java index 3af90ea9a0a1..9cce1f71f2a1 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java @@ -336,35 +336,6 @@ public void output(OutputT output, Instant timestamp, BoundedWindow window) { public void output(TupleTag tag, T output, Instant timestamp, BoundedWindow window) { outputWindowedValue(tag, WindowedValues.of(output, timestamp, window, PaneInfo.NO_FIRING)); } - - @Override - public void output( - OutputT output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - output(mainOutputTag, output, timestamp, window, currentRecordId, currentRecordOffset); - } - - @Override - public void output( - TupleTag tag, - T output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - outputWindowedValue( - tag, - WindowedValues.of( - output, - timestamp, - Collections.singletonList(window), - PaneInfo.NO_FIRING, - currentRecordId, - currentRecordOffset)); - } } private final DoFnFinishBundleArgumentProvider.Context context = @@ -461,24 +432,6 @@ public void outputWindowedValue( outputWindowedValue(mainOutputTag, output, timestamp, windows, paneInfo); } - @Override - public void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - outputWindowedValue( - mainOutputTag, - output, - timestamp, - windows, - paneInfo, - currentRecordId, - currentRecordOffset); - } - @Override public void output(TupleTag tag, T output) { checkNotNull(tag, "Tag passed to output cannot be null"); @@ -512,21 +465,6 @@ public void outputWindowedValue( .output(); } - @Override - public void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - SimpleDoFnRunner.this.outputWindowedValue( - tag, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } - @Override public Instant timestamp() { return elem.getTimestamp(); @@ -964,24 +902,6 @@ public void outputWindowedValue( outputWindowedValue(mainOutputTag, output, timestamp, windows, paneInfo); } - @Override - public void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - outputWindowedValue( - mainOutputTag, - output, - timestamp, - windows, - paneInfo, - currentRecordId, - currentRecordOffset); - } - @Override public void output(TupleTag tag, T output) { checkTimestamp(timestamp(), timestamp); @@ -1013,22 +933,6 @@ public void outputWindowedValue( .output(); } - @Override - public void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - checkTimestamp(timestamp(), timestamp); - SimpleDoFnRunner.this.outputWindowedValue( - tag, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } - @Override public BundleFinalizer bundleFinalizer() { throw new UnsupportedOperationException( @@ -1243,24 +1147,6 @@ public void outputWindowedValue( outputWindowedValue(mainOutputTag, output, timestamp, windows, paneInfo); } - @Override - public void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - outputWindowedValue( - mainOutputTag, - output, - timestamp, - windows, - paneInfo, - currentRecordId, - currentRecordOffset); - } - @Override public void output(TupleTag tag, T output) { checkTimestamp(this.timestamp, timestamp); @@ -1291,22 +1177,6 @@ public void outputWindowedValue( .output(); } - @Override - public void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - checkTimestamp(this.timestamp, timestamp); - SimpleDoFnRunner.this.outputWindowedValue( - tag, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } - @Override public BundleFinalizer bundleFinalizer() { throw new UnsupportedOperationException( diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDoViaKeyedWorkItems.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDoViaKeyedWorkItems.java index 6af54da0a08b..9cf6db23f244 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDoViaKeyedWorkItems.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDoViaKeyedWorkItems.java @@ -662,27 +662,6 @@ public void output( throwUnsupportedOutput(); } - @Override - public void output( - OutputT output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - throwUnsupportedOutput(); - } - - @Override - public void output( - TupleTag tag, - T output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - throwUnsupportedOutput(); - } - @Override public PipelineOptions getPipelineOptions() { return baseContext.getPipelineOptions(); diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java index d0714de60328..a82e84090cb7 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java @@ -123,12 +123,6 @@ public abstract class FinishBundleContext { */ public abstract void output(OutputT output, Instant timestamp, BoundedWindow window); - public abstract void output( - OutputT output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset); /** * Adds the given element to the output {@code PCollection} with the given tag at the given * timestamp in the given window. @@ -140,14 +134,6 @@ public abstract void output( */ public abstract void output( TupleTag tag, T output, Instant timestamp, BoundedWindow window); - - public abstract void output( - TupleTag tag, - T output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset); } /** @@ -226,14 +212,6 @@ public abstract void outputWindowedValue( Collection windows, PaneInfo paneInfo); - public abstract void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset); - /** * Adds the given element to the output {@code PCollection} with the given tag. * @@ -306,15 +284,6 @@ public abstract void outputWindowedValue( Instant timestamp, Collection windows, PaneInfo paneInfo); - - public abstract void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset); } /** Information accessible when running a {@link DoFn.ProcessElement} method. */ diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java index c59d6b528c3f..3bdeb57ed888 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java @@ -492,35 +492,6 @@ public void output(TupleTag tag, T output, Instant timestamp, BoundedWind ValueInSingleWindow.of( output, timestamp, window, PaneInfo.NO_FIRING, null, null)); } - - @Override - public void output( - OutputT output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - output(mainOutputTag, output, timestamp, window, currentRecordId, currentRecordOffset); - } - - @Override - public void output( - TupleTag tag, - T output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - getMutableOutput(tag) - .add( - ValueInSingleWindow.of( - output, - timestamp, - window, - PaneInfo.NO_FIRING, - currentRecordId, - currentRecordOffset)); - } }; } @@ -642,24 +613,6 @@ public void outputWindowedValue( outputWindowedValue(mainOutputTag, output, timestamp, windows, paneInfo); } - @Override - public void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - outputWindowedValue( - mainOutputTag, - output, - timestamp, - windows, - paneInfo, - currentRecordId, - currentRecordOffset); - } - @Override public void output(TupleTag tag, T output) { outputWithTimestamp(tag, output, element.getTimestamp()); @@ -685,23 +638,6 @@ public void outputWindowedValue( .add(ValueInSingleWindow.of(output, timestamp, w, paneInfo, null, null)); } } - - @Override - public void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - for (BoundedWindow w : windows) { - getMutableOutput(tag) - .add( - ValueInSingleWindow.of( - output, timestamp, w, paneInfo, currentRecordId, currentRecordOffset)); - } - } } /** @deprecated Use {@link TestPipeline} with the {@code DirectRunner}. */ diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/SplittableParDoNaiveBounded.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/SplittableParDoNaiveBounded.java index e6394b8810a4..9f5322fb5116 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/SplittableParDoNaiveBounded.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/SplittableParDoNaiveBounded.java @@ -416,29 +416,6 @@ public void output( "Output from FinishBundle for SDF is not supported in naive implementation"); } - @Override - public void output( - TupleTag tag, - T output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - throw new UnsupportedOperationException( - "Output from FinishBundle for SDF is not supported in naive implementation"); - } - - @Override - public void output( - @Nullable OutputT output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - throw new UnsupportedOperationException( - "Output from FinishBundle for SDF is not supported in naive implementation"); - } - @Override public void output( TupleTag tag, T output, Instant timestamp, BoundedWindow window) { @@ -651,18 +628,6 @@ public void outputWindowedValue( outerContext.outputWindowedValue(output, timestamp, windows, paneInfo); } - @Override - public void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - outerContext.outputWindowedValue( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset); - } - @Override public void output(TupleTag tag, T output) { outerContext.output(tag, output); @@ -683,19 +648,6 @@ public void outputWindowedValue( outerContext.outputWindowedValue(tag, output, timestamp, windows, paneInfo); } - @Override - public void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - outerContext.outputWindowedValue( - tag, output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset); - } - @Override public InputT element() { return element; diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java index 0388d3c03f00..1b7d75f6ec32 100644 --- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java +++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java @@ -1668,48 +1668,6 @@ public void output(TupleTag tag, T output, Instant timestamp, BoundedWind } outputTo(consumer, WindowedValues.of(output, timestamp, window, PaneInfo.NO_FIRING)); } - - @Override - public void output( - OutputT output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - outputTo( - mainOutputConsumer, - WindowedValues.of( - output, - timestamp, - Collections.singletonList(window), - PaneInfo.NO_FIRING, - currentRecordId, - currentRecordOffset)); - } - - @Override - public void output( - TupleTag tag, - T output, - Instant timestamp, - BoundedWindow window, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - FnDataReceiver> consumer = - (FnDataReceiver) localNameToConsumer.get(tag.getId()); - if (consumer == null) { - throw new IllegalArgumentException(String.format("Unknown output tag %s", tag)); - } - outputTo( - consumer, - WindowedValues.of( - output, - timestamp, - Collections.singletonList(window), - PaneInfo.NO_FIRING, - currentRecordId, - currentRecordOffset)); - } } private final FinishBundleArgumentProvider.Context context = @@ -1808,22 +1766,6 @@ public void outputWindowedValue( outputTo(mainOutputConsumer, WindowedValues.of(output, timestamp, windows, paneInfo)); } - @Override - public void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - // TODO(https://github.com/apache/beam/issues/29637): Check that timestamp is valid once all - // runners can provide proper timestamps. - outputTo( - mainOutputConsumer, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } - @Override public void outputWithTimestamp(TupleTag tag, T output, Instant timestamp) { // TODO(https://github.com/apache/beam/issues/29637): Check that timestamp is valid once all @@ -1855,26 +1797,6 @@ public void outputWindowedValue( outputTo(consumer, WindowedValues.of(output, timestamp, windows, paneInfo)); } - @Override - public void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - FnDataReceiver> consumer = - (FnDataReceiver) localNameToConsumer.get(tag.getId()); - if (consumer == null) { - throw new IllegalArgumentException(String.format("Unknown output tag %s", tag)); - } - outputTo( - consumer, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } - @Override public State state(String stateId, boolean alwaysFetched) { StateDeclaration stateDeclaration = doFnSignature.stateDeclarations().get(stateId); @@ -1978,21 +1900,6 @@ public void outputWindowedValue( builder(output).setTimestamp(timestamp).setWindows(windows).setPaneInfo(paneInfo).output(); } - @Override - public void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - checkTimestamp(timestamp); - outputTo( - mainOutputConsumer, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } - @Override public void outputWithTimestamp(TupleTag tag, T output, Instant timestamp) { checkTimestamp(timestamp); @@ -2022,27 +1929,6 @@ public void outputWindowedValue( } outputTo(consumer, WindowedValues.of(output, timestamp, windows, paneInfo)); } - - @Override - public void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - checkTimestamp(timestamp); - FnDataReceiver> consumer = - (FnDataReceiver) localNameToConsumer.get(tag.getId()); - if (consumer == null) { - throw new IllegalArgumentException(String.format("Unknown output tag %s", tag)); - } - outputTo( - consumer, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } } /** Provides base arguments for a {@link DoFnInvoker} for a non-window observing method. */ @@ -2399,21 +2285,6 @@ public OutputBuilder builder(OutputT value) { }); } - @Override - public void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - checkOnWindowExpirationTimestamp(timestamp); - outputTo( - mainOutputConsumer, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } - @Override public void output(TupleTag tag, T output) { FnDataReceiver> consumer = @@ -2450,25 +2321,10 @@ public void outputWindowedValue( Instant timestamp, Collection windows, PaneInfo paneInfo) { - outputWindowedValue(tag, output, timestamp, windows, paneInfo, null, null); - } - - @Override - public void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { checkOnWindowExpirationTimestamp(timestamp); FnDataReceiver> consumer = (FnDataReceiver) localNameToConsumer.get(tag.getId()); - outputTo( - consumer, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); + outputTo(consumer, WindowedValues.of(output, timestamp, windows, paneInfo)); } @SuppressWarnings( @@ -2723,21 +2579,6 @@ public void outputWindowedValue( OutputReceiver.super.outputWindowedValue(output, timestamp, windows, paneInfo); } - @Override - public void outputWindowedValue( - OutputT output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) { - checkTimerTimestamp(timestamp); - outputTo( - mainOutputConsumer, - WindowedValues.of( - output, timestamp, windows, paneInfo, currentRecordId, currentRecordOffset)); - } - @Override public void output(TupleTag tag, T output) { checkTimerTimestamp(currentTimer.getHoldTimestamp()); @@ -2776,16 +2617,6 @@ public void outputWindowedValue( Collection windows, PaneInfo paneInfo) {} - @Override - public void outputWindowedValue( - TupleTag tag, - T output, - Instant timestamp, - Collection windows, - PaneInfo paneInfo, - @Nullable String currentRecordId, - @Nullable Long currentRecordOffset) {} - @Override public TimeDomain timeDomain() { return currentTimeDomain; diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java index c342da810e0c..568fe49217b3 100644 --- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java +++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java @@ -94,7 +94,6 @@ import org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators.Manual; import org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators.MonotonicallyIncreasing; import org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators.WallTime; -import org.apache.beam.sdk.transforms.windowing.GlobalWindow; import org.apache.beam.sdk.util.Preconditions; import org.apache.beam.sdk.util.construction.PTransformMatchers; import org.apache.beam.sdk.util.construction.ReplacementOutputs; @@ -112,7 +111,6 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Joiner; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; @@ -2016,8 +2014,8 @@ static class OffsetDeduplicationIdExtractor extends DoFn, KafkaRecord> { @ProcessElement - public void processElement(ProcessContext pc) { - KafkaRecord element = pc.element(); + public void processElement( + @Element KafkaRecord element, OutputReceiver> outputReceiver) { Long offset = null; String uniqueId = null; if (element != null) { @@ -2025,13 +2023,7 @@ public void processElement(ProcessContext pc) { uniqueId = (String.format("%s-%d-%d", element.getTopic(), element.getPartition(), offset)); } - pc.outputWindowedValue( - element, - pc.timestamp(), - Lists.newArrayList(GlobalWindow.INSTANCE), - pc.pane(), - uniqueId, - offset); + outputReceiver.builder(element).setRecordId(uniqueId).setRecordOffset(offset).output(); } }