Enable SDF testPairWithIndexWindowedTimestampedBounded in Spark Structured Streaming runner

[iemejia]

This passes now after some recent changes in core. A really minor improvement after reviewing the PR on SDF Bounded reads for Spark runner

R: @lukecwik

[iemejia]

Run Spark StructuredStreaming ValidatesRunner

[iemejia]

Please remember than there are two GbK tests failing on the VR suite for this runner, so this is just to validate that the now unexcluded test passes correctly.

[iemejia]

@lukecwik It seems there is a new test failing on the Structured Streaming runner after we merged the SDF read by default change. Not sure if it could be related, do you see any relation?

https://scans.gradle.com/s/lsf5y44b36pyc/tests/:runners:spark:validatesStructuredStreamingRunnerBatch/org.apache.beam.sdk.transforms.CombineTest$WindowingTests/testSlidingWindowsCombine#1

[lukecwik]

There seems to be a bug in the AggregatorCombiner:

The test passes occassionaly and it depends on the order of merge/reduce steps that Spark does. A good run:

LCWIK merge accum1: [] accum2: [TimestampedValueInMultipleWindows{value=[c], timestamp=1970-01-01T00:00:00.003Z, windows=[[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), [1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), [1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}}] rval: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}]
LCWIK merge accum1: [] accum2: [TimestampedValueInMultipleWindows{value=[b], timestamp=1970-01-01T00:00:00.002Z, windows=[[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), [1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), [1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}}] rval: [TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK merge accum1: [] accum2: [TimestampedValueInMultipleWindows{value=[a], timestamp=1970-01-01T00:00:00.001Z, windows=[[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), [1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), [1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}}] rval: [TimestampedValueInSingleWindow{value=[a], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK reduce value: TimestampedValueInMultipleWindows{value=KV{null, b}, timestamp=1970-01-01T00:00:00.002Z, windows=[[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), [1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), [1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}} accum: [] result: [TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK reduce value: TimestampedValueInMultipleWindows{value=KV{null, a}, timestamp=1970-01-01T00:00:00.001Z, windows=[[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), [1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), [1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}} accum: [] result: [TimestampedValueInSingleWindow{value=[a], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK reduce value: TimestampedValueInMultipleWindows{value=KV{null, c}, timestamp=1970-01-01T00:00:00.003Z, windows=[[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), [1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), [1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}} accum: [] result: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}]
LCWIK merge accum1: [] accum2: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}] rval: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}]
LCWIK merge accum1: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}] accum2: [TimestampedValueInSingleWindow{value=[b, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}] rval: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK merge accum1: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}] accum2: [TimestampedValueInSingleWindow{value=[a], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, b, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, b], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}] rval: [TimestampedValueInSingleWindow{value=[a], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, b, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, b], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]

Bad run:

LCWIK merge accum1: [] accum2: [TimestampedValueInMultipleWindows{value=[c], timestamp=1970-01-01T00:00:00.003Z, windows=[[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), [1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), [1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}}] rval: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}]
LCWIK merge accum1: [] accum2: [TimestampedValueInMultipleWindows{value=[b], timestamp=1970-01-01T00:00:00.002Z, windows=[[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), [1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), [1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}}] rval: [TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK reduce value: TimestampedValueInMultipleWindows{value=KV{null, c}, timestamp=1970-01-01T00:00:00.003Z, windows=[[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), [1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), [1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}} accum: [] result: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}]
LCWIK reduce value: TimestampedValueInMultipleWindows{value=KV{null, b}, timestamp=1970-01-01T00:00:00.002Z, windows=[[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), [1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), [1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}} accum: [] result: [TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK merge accum1: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}] accum2: [TimestampedValueInMultipleWindows{value=[a, c], timestamp=1970-01-01T00:00:00.001Z, windows=[[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), [1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), [1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}}] rval: [TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK reduce value: TimestampedValueInMultipleWindows{value=KV{null, a}, timestamp=1970-01-01T00:00:00.001Z, windows=[[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), [1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), [1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z)], pane=PaneInfo{isFirst=true, timing=EARLY, index=0}} accum: [TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}] result: [TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK merge accum1: [] accum2: [TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}] rval: [TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]
LCWIK merge accum1: [TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}] accum2: [TimestampedValueInSingleWindow{value=[b, a, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, a, c], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}] rval: [TimestampedValueInSingleWindow{value=[a, c], timestamp=1970-01-01T00:00:00.001Z, window=[1969-12-31T23:59:59.999Z..1970-01-01T00:00:00.002Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[c], timestamp=1970-01-01T00:00:00.005Z, window=[1970-01-01T00:00:00.003Z..1970-01-01T00:00:00.006Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, a, c], timestamp=1970-01-01T00:00:00.003Z, window=[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.004Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, c], timestamp=1970-01-01T00:00:00.004Z, window=[1970-01-01T00:00:00.002Z..1970-01-01T00:00:00.005Z), pane=PaneInfo.NO_FIRING}, TimestampedValueInSingleWindow{value=[b, a, c], timestamp=1970-01-01T00:00:00.002Z, window=[1970-01-01T00:00:00.000Z..1970-01-01T00:00:00.003Z), pane=PaneInfo.NO_FIRING}]

[lukecwik]

@iemejia The issue was that the AggregatorCombiner was using the same accumulator instance across multiple windows which lead to it being mutated across windows during merging. Filed BEAM-11050 with fix in #13061

[iemejia]

Excellent thanks a lot for quickly finding the issue (and for the fix).