[ https://issues.apache.org/jira/browse/BEAM-3776?focusedWorklogId=82064&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-82064 ]
ASF GitHub Bot logged work on BEAM-3776: ---------------------------------------- Author: ASF GitHub Bot Created on: 19/Mar/18 22:15 Start Date: 19/Mar/18 22:15 Worklog Time Spent: 10m Work Description: tgroh commented on a change in pull request #4793: [BEAM-3776] Fix issue with merging late windows where a watermark hold could be added behind the input watermark. URL: https://github.com/apache/beam/pull/4793#discussion_r175586984 ########## File path: runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnRunnerTest.java ########## @@ -873,6 +911,288 @@ public void testWatermarkHoldAndLateData() throws Exception { tester.assertHasOnlyGlobalAndFinishedSetsFor(); } + // Performs the specified actions and verifies that the watermark hold is set correctly. + public void mergingWatermarkHoldTestHelper(List<Action> configuration) throws Exception { + LOG.info("Running config {}", configuration); + MetricsContainerImpl container = new MetricsContainerImpl("any"); + MetricsEnvironment.setCurrentContainer(container); + // Test handling of late data. Specifically, ensure the watermark hold is correct. + Duration allowedLateness = Duration.standardMinutes(1); + Duration gapDuration = Duration.millis(10); + LOG.info("Gap duration {}", gapDuration); + ReduceFnTester<Integer, Iterable<Integer>, IntervalWindow> tester = + ReduceFnTester.nonCombining( + WindowingStrategy.of(Sessions.withGapDuration(gapDuration)) + .withMode(AccumulationMode.DISCARDING_FIRED_PANES) + .withTrigger( + Repeatedly.forever( + AfterWatermark.pastEndOfWindow() + .withLateFirings(AfterPane.elementCountAtLeast(1)))) + .withAllowedLateness(allowedLateness)); + tester.setAutoAdvanceOutputWatermark(false); + + // Input watermark -> null + assertEquals(null, tester.getWatermarkHold()); + assertEquals(null, tester.getOutputWatermark()); + + int maxTs = 0; + long watermark = 0; + for (Action action : configuration) { + if (action.times != null) { + LOG.info("Injecting {}", action.times); + injectElements(tester, action.times); + int maxLocalTs = Ordering.natural().max(action.times); + if (maxLocalTs > maxTs) { + maxTs = maxLocalTs; + } + } + if (action.inputWatermark > watermark) { + watermark = action.inputWatermark; + LOG.info("Advancing watermark to {}", new Instant(watermark)); + tester.advanceInputWatermark(new Instant(watermark)); + } + Instant hold = tester.getWatermarkHold(); + if (hold != null) { + assertThat(hold, greaterThanOrEqualTo(new Instant(watermark))); + assertThat(watermark, lessThan(maxTs + gapDuration.getMillis())); + } + } + if (gapDuration.getMillis() + maxTs > watermark) { + watermark = gapDuration.getMillis() + maxTs; + tester.advanceInputWatermark(new Instant(watermark)); + } + LOG.info("Output {}", tester.extractOutput()); + assertThat(tester.getWatermarkHold(), nullValue()); + tester.advanceInputWatermark(new Instant(watermark).plus(allowedLateness)); + assertThat(tester.getWatermarkHold(), nullValue()); + + // Nothing dropped. + long droppedElements = + container + .getCounter( + MetricName.named(ReduceFnRunner.class, + ReduceFnRunner.DROPPED_DUE_TO_CLOSED_WINDOW)) + .getCumulative() + .longValue(); + assertEquals(0, droppedElements); + } + + @Test + public void testMergingWatermarkHoldLateNewWindow() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(1)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingWatermarkHoldLateNewWindowExtended() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(1)); + actions.add(Action.times(10)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingWatermarkHoldLateNewWindowMerged() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(1)); + actions.add(Action.times(14)); + actions.add(Action.times(6)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingWatermarkHoldLateNewWindowExtendedPastInputWatermark() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(25)); + actions.add(Action.times(33)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingWatermarkHoldLateNewWindowExtendedPastInputWatermarkAndFurther() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(25)); + actions.add(Action.times(33)); + actions.add(Action.times(43)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingWatermarkHoldExtendClosedWindow() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.times(11)); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(18)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingWatermarkHoldExtendClosedWindowAndMergeWithNew() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.times(11)); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(18)); + actions.add(Action.times(41)); + actions.add(Action.times(27, 33)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingWatermarkHoldLateNewWindowMergedWithOntime() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(29)); + actions.add(Action.times(45)); + actions.add(Action.times(36)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingWatermarkHoldLateNewWindowMergedWithOntimeSingleProcessing() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(29, 45, 36)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingWatermarkHoldRepeatedMerging() throws Exception { + LinkedList<Action> actions = new LinkedList<>(); + actions.add(Action.inputWatermark(40)); + actions.add(Action.times(25)); + actions.add(Action.times(42)); + actions.add(Action.times(33, 21)); + actions.add(Action.inputWatermark(50)); + actions.add(Action.times(12)); + mergingWatermarkHoldTestHelper(actions); + } + + @Test + public void testMergingLateWatermarkHolds() throws Exception { + MetricsContainerImpl container = new MetricsContainerImpl("any"); + MetricsEnvironment.setCurrentContainer(container); + Duration gapDuration = Duration.millis(10); + Duration allowedLateness = Duration.standardMinutes(100); + ReduceFnTester<Integer, Iterable<Integer>, IntervalWindow> tester = + ReduceFnTester.nonCombining( + WindowingStrategy.of(Sessions.withGapDuration(gapDuration)) + .withMode(AccumulationMode.DISCARDING_FIRED_PANES) + .withTrigger( + Repeatedly.forever( + AfterWatermark.pastEndOfWindow() + .withLateFirings(AfterPane.elementCountAtLeast(10)))) + .withAllowedLateness(allowedLateness)); + tester.setAutoAdvanceOutputWatermark(false); + + // Input watermark -> null + assertEquals(null, tester.getWatermarkHold()); + assertEquals(null, tester.getOutputWatermark()); + + tester.advanceInputWatermark(new Instant(20)); + // Add two late elements that cause a window to merge. + injectElements(tester, Arrays.asList(3)); + assertThat(tester.getWatermarkHold(), nullValue()); + injectElements(tester, Arrays.asList(4)); + Instant endOfWindow = new Instant(4).plus(gapDuration); + // We expect a GC hold to be one less than the end of window plus the allowed lateness. + Instant expectedGcHold = endOfWindow.plus(allowedLateness).minus(1); + assertEquals( + expectedGcHold, + tester.getWatermarkHold()); + tester.advanceInputWatermark(new Instant(1000)); + assertEquals( + expectedGcHold, + tester.getWatermarkHold()); + } + + @Test + public void testMergingWatermarkHoldAndLateDataFuzz() throws Exception { + MetricsContainerImpl container = new MetricsContainerImpl("any"); + MetricsEnvironment.setCurrentContainer(container); + // Test handling of late data. Specifically, ensure the watermark hold is correct. + Duration allowedLateness = Duration.standardMinutes(100); + ThreadLocalRandom r = ThreadLocalRandom.current(); Review comment: If you're going to do this, use it to seed a new `Random` with the seed logged, so we can reproduce any flakes. I'm not huge on randomness in tests, but I'm especially not huge on non-reproducible randomness in tests. Make sure that an SLF4J logger is on the test classpath, and probably that it's `slf4j_simple` ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking ------------------- Worklog Id: (was: 82064) > StateMerging.mergeWatermarks sets a late watermark hold for late merging > windows that depend only on the window > --------------------------------------------------------------------------------------------------------------- > > Key: BEAM-3776 > URL: https://issues.apache.org/jira/browse/BEAM-3776 > Project: Beam > Issue Type: Bug > Components: runner-core > Affects Versions: 2.1.0, 2.2.0, 2.3.0 > Reporter: Sam Whittle > Assignee: Sam Whittle > Priority: Critical > Time Spent: 2h 50m > Remaining Estimate: 0h > > WatermarkHold.addElementHold and WatermarkHold.addGarbageCollectionHold take > to not add holds that would be before the input watermark. > However WatermarkHold.onMerge calls StateMerging.mergeWatermarks which if the > window depends only on window, sets a hold for the end of the window > regardless of the input watermark. > Thus if you have a WindowingStrategy such as: > WindowingStrategy.of(Sessions.withGapDuration(gapDuration)) > .withMode(AccumulationMode.DISCARDING_FIRED_PANES) > .withTrigger( > Repeatedly.forever( > AfterWatermark.pastEndOfWindow() > .withLateFirings(AfterPane.elementCountAtLeast(10)))) > .withAllowedLateness(allowedLateness)) > and you merge windows that are late, you might end up holding the watermark > until the allowedLateness has passed. -- This message was sent by Atlassian JIRA (v7.6.3#76005)