From f5582f74c8e41d7530dd947c74c1426962833e5f Mon Sep 17 00:00:00 2001 From: vishal Date: Wed, 23 Sep 2026 21:43:37 +0530 Subject: [PATCH 1/2] Fix: Prevent CommitTooLargeException in OrderedEventProcessor on duplicate floods --- .../ordered/GlobalSequencesProcessorDoFn.java | 2 + .../extensions/ordered/ProcessingState.java | 4 ++ .../sdk/extensions/ordered/ProcessorDoFn.java | 20 +++++--- .../ordered/SequencePerKeyProcessorDoFn.java | 2 + ...deredEventProcessorPerKeySequenceTest.java | 48 +++++++++++++++++++ 5 files changed, 69 insertions(+), 7 deletions(-) diff --git a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequencesProcessorDoFn.java b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequencesProcessorDoFn.java index 3c6c72eb486b..0f93329407b2 100644 --- a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequencesProcessorDoFn.java +++ b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequencesProcessorDoFn.java @@ -175,6 +175,7 @@ public void processElement( if (numberOfResultsBeforeBundleStart == null) { // Per key processing is synchronized by Beam. There is no need to have it here. numberOfResultsBeforeBundleStart = processingState.getResultCount(); + numberOfDuplicatesBeforeBundleStart = processingState.getDuplicates(); } processingState.eventReceived(); @@ -262,6 +263,7 @@ public void onBatchEmission( } this.numberOfResultsBeforeBundleStart = processingState.getResultCount(); + this.numberOfDuplicatesBeforeBundleStart = processingState.getDuplicates(); state = processBufferedEventRange( diff --git a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessingState.java b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessingState.java index 2980c2d614f0..f7c8490de6c0 100644 --- a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessingState.java +++ b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessingState.java @@ -313,6 +313,10 @@ public long resultsProducedInBundle(long numberOfResultsBeforeBundleStart) { return resultCount - numberOfResultsBeforeBundleStart; } + public long duplicatesProducedInBundle(long numberOfDuplicatesBeforeBundleStart) { + return duplicates - numberOfDuplicatesBeforeBundleStart; + } + public void updateGlobalSequenceDetails(ContiguousSequenceRange updated) { if (thereAreGloballySequencedEventsToBeProcessed()) { // We don't update the timer if we can already process events in the onTimer batch. diff --git a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessorDoFn.java b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessorDoFn.java index 3e97f85cd596..94316e64dd88 100644 --- a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessorDoFn.java +++ b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessorDoFn.java @@ -63,6 +63,7 @@ abstract class ProcessorDoFn< private final long maxNumberOfResultsToProduce; protected @Nullable Long numberOfResultsBeforeBundleStart = 0L; + protected @Nullable Long numberOfDuplicatesBeforeBundleStart = 0L; ProcessorDoFn( EventExaminer eventExaminer, @@ -85,12 +86,14 @@ abstract class ProcessorDoFn< @StartBundle public void onBundleStart() { numberOfResultsBeforeBundleStart = null; + numberOfDuplicatesBeforeBundleStart = null; } @FinishBundle public void onBundleFinish() { // This might be necessary because this field is also used in a Timer numberOfResultsBeforeBundleStart = null; + numberOfDuplicatesBeforeBundleStart = null; } /** Returns true if each event needs to be examined. */ @@ -267,10 +270,12 @@ protected void emitProcessingStatus( protected boolean reachedMaxResultCountForBundle( ProcessingState processingState, Timer largeBatchEmissionTimer) { - boolean exceeded = - processingState.resultsProducedInBundle( - numberOfResultsBeforeBundleStart == null ? 0 : numberOfResultsBeforeBundleStart) - >= maxNumberOfResultsToProduce; + long resultsEmitted = processingState.resultsProducedInBundle( + numberOfResultsBeforeBundleStart == null ? 0 : numberOfResultsBeforeBundleStart); + long duplicatesEmitted = processingState.duplicatesProducedInBundle( + numberOfDuplicatesBeforeBundleStart == null ? 0 : numberOfDuplicatesBeforeBundleStart); + + boolean exceeded = (resultsEmitted + duplicatesEmitted) >= maxNumberOfResultsToProduce; if (exceeded) { if (LOG.isTraceEnabled()) { LOG.trace( @@ -354,9 +359,10 @@ StateT processBufferedEventRange( beforeInitialSequence ? Reason.before_initial_sequence : Reason.duplicate)))); - // TODO: When there is a large number of duplicates this can cause a situation where - // we produce too much output and the runner will start throwing unrecoverable errors. - // Need to add counting logic to accumulate both the normal and DLQ outputs. + if (reachedMaxResultCountForBundle(processingState, largeBatchEmissionTimer)) { + endClearRange = fromLong(eventSequence + 1); + break; + } continue; } diff --git a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/SequencePerKeyProcessorDoFn.java b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/SequencePerKeyProcessorDoFn.java index 486622d2069a..d42e84079fda 100644 --- a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/SequencePerKeyProcessorDoFn.java +++ b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/SequencePerKeyProcessorDoFn.java @@ -159,6 +159,7 @@ public void processElement( if (numberOfResultsBeforeBundleStart == null) { // Per key processing is synchronized by Beam. There is no need to have it here. numberOfResultsBeforeBundleStart = processingState.getResultCount(); + numberOfDuplicatesBeforeBundleStart = processingState.getDuplicates(); } processingState.eventReceived(); @@ -259,6 +260,7 @@ public void onBatchEmission( LOG.debug("Starting to process batch for key '{}'", processingState.getKey()); this.numberOfResultsBeforeBundleStart = processingState.getResultCount(); + this.numberOfDuplicatesBeforeBundleStart = processingState.getDuplicates(); processBufferedEvents( processingState, state, bufferedEventsState, outputReceiver, largeBatchEmissionTimer); diff --git a/sdks/java/extensions/ordered/src/test/java/org/apache/beam/sdk/extensions/ordered/OrderedEventProcessorPerKeySequenceTest.java b/sdks/java/extensions/ordered/src/test/java/org/apache/beam/sdk/extensions/ordered/OrderedEventProcessorPerKeySequenceTest.java index 5dad7ac1852b..c0b5cfb4d1f4 100644 --- a/sdks/java/extensions/ordered/src/test/java/org/apache/beam/sdk/extensions/ordered/OrderedEventProcessorPerKeySequenceTest.java +++ b/sdks/java/extensions/ordered/src/test/java/org/apache/beam/sdk/extensions/ordered/OrderedEventProcessorPerKeySequenceTest.java @@ -246,6 +246,54 @@ public void testHandlingOfDuplicateSequences() throws CannotProvideCoderExceptio DONT_PRODUCE_STATUS_ON_EVERY_EVENT); } + @Test + public void testLargeNumberOfDuplicatesPaginatesCorrectly() throws CannotProvideCoderException { + int maxResultsPerOutput = 10; + int duplicateCount = 25; + List events = new ArrayList<>(); + events.add(Event.create(0, "id-1", "a")); + + // Add 25 duplicates of the next event. The maxResultsPerOutput is 10, so it should + // paginate the duplicates across multiple bundles. + for (int i = 0; i < duplicateCount; i++) { + events.add(Event.create(1, "id-1", "b")); + } + + Collection> expectedStatuses = new ArrayList<>(); + expectedStatuses.add( + KV.of( + "id-1", + OrderedProcessingStatus.create( + 1L, + 0, + null, + null, + events.size(), + 2L, + duplicateCount - 1, // one is processed, the rest are duplicates + false, + NOT_USED_FOR_TESTING))); + + Collection> expectedOutput = new ArrayList<>(); + expectedOutput.add(KV.of("id-1", "a")); + expectedOutput.add(KV.of("id-1", "ab")); + + Collection>>> duplicates = new ArrayList<>(); + for (int i = 0; i < duplicateCount - 1; i++) { + duplicates.add(KV.of("id-1", KV.of(1L, UnprocessedEvent.create("b", Reason.duplicate)))); + } + + testPerKeySequenceProcessing( + events.toArray(new Event[0]), + expectedStatuses, + expectedOutput, + duplicates, + EMISSION_FREQUENCY_ON_EVERY_ELEMENT, + INITIAL_SEQUENCE_OF_0, + maxResultsPerOutput, + DONT_PRODUCE_STATUS_ON_EVERY_EVENT); + } + @Test public void testHandlingOfCheckedExceptions() throws CannotProvideCoderException { Event[] events = { From f2bf550d8ab09ccbdde0aaaf227a806e1f450297 Mon Sep 17 00:00:00 2001 From: vishal Date: Fri, 25 Sep 2026 19:45:02 +0530 Subject: [PATCH 2/2] Apply spotless formatting fixes --- .../beam/sdk/extensions/ordered/ProcessorDoFn.java | 10 ++++++---- .../OrderedEventProcessorPerKeySequenceTest.java | 6 +++--- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessorDoFn.java b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessorDoFn.java index 94316e64dd88..ead50556697d 100644 --- a/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessorDoFn.java +++ b/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ProcessorDoFn.java @@ -270,10 +270,12 @@ protected void emitProcessingStatus( protected boolean reachedMaxResultCountForBundle( ProcessingState processingState, Timer largeBatchEmissionTimer) { - long resultsEmitted = processingState.resultsProducedInBundle( - numberOfResultsBeforeBundleStart == null ? 0 : numberOfResultsBeforeBundleStart); - long duplicatesEmitted = processingState.duplicatesProducedInBundle( - numberOfDuplicatesBeforeBundleStart == null ? 0 : numberOfDuplicatesBeforeBundleStart); + long resultsEmitted = + processingState.resultsProducedInBundle( + numberOfResultsBeforeBundleStart == null ? 0 : numberOfResultsBeforeBundleStart); + long duplicatesEmitted = + processingState.duplicatesProducedInBundle( + numberOfDuplicatesBeforeBundleStart == null ? 0 : numberOfDuplicatesBeforeBundleStart); boolean exceeded = (resultsEmitted + duplicatesEmitted) >= maxNumberOfResultsToProduce; if (exceeded) { diff --git a/sdks/java/extensions/ordered/src/test/java/org/apache/beam/sdk/extensions/ordered/OrderedEventProcessorPerKeySequenceTest.java b/sdks/java/extensions/ordered/src/test/java/org/apache/beam/sdk/extensions/ordered/OrderedEventProcessorPerKeySequenceTest.java index c0b5cfb4d1f4..45a4a891adc9 100644 --- a/sdks/java/extensions/ordered/src/test/java/org/apache/beam/sdk/extensions/ordered/OrderedEventProcessorPerKeySequenceTest.java +++ b/sdks/java/extensions/ordered/src/test/java/org/apache/beam/sdk/extensions/ordered/OrderedEventProcessorPerKeySequenceTest.java @@ -252,11 +252,11 @@ public void testLargeNumberOfDuplicatesPaginatesCorrectly() throws CannotProvide int duplicateCount = 25; List events = new ArrayList<>(); events.add(Event.create(0, "id-1", "a")); - + // Add 25 duplicates of the next event. The maxResultsPerOutput is 10, so it should // paginate the duplicates across multiple bundles. for (int i = 0; i < duplicateCount; i++) { - events.add(Event.create(1, "id-1", "b")); + events.add(Event.create(1, "id-1", "b")); } Collection> expectedStatuses = new ArrayList<>(); @@ -280,7 +280,7 @@ public void testLargeNumberOfDuplicatesPaginatesCorrectly() throws CannotProvide Collection>>> duplicates = new ArrayList<>(); for (int i = 0; i < duplicateCount - 1; i++) { - duplicates.add(KV.of("id-1", KV.of(1L, UnprocessedEvent.create("b", Reason.duplicate)))); + duplicates.add(KV.of("id-1", KV.of(1L, UnprocessedEvent.create("b", Reason.duplicate)))); } testPerKeySequenceProcessing(