[jira] [Commented] (BEAM-3225) Non deterministic behaviour of AfterProcessingTime trigger with multiple group by transformations
[
https://issues.apache.org/jira/browse/BEAM-3225?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16403089#comment-16403089
]
Pawel Bartoszek commented on BEAM-3225:
---
[~dawidwys] I think this issue might have been caused by
https://issues.apache.org/jira/browse/BEAM-3863
> Non deterministic behaviour of AfterProcessingTime trigger with multiple
> group by transformations
> -
>
> Key: BEAM-3225
> URL: https://issues.apache.org/jira/browse/BEAM-3225
> Project: Beam
> Issue Type: Bug
> Components: runner-core, runner-flink
>Affects Versions: 2.0.0, 2.1.0
>Reporter: Pawel Bartoszek
>Assignee: Aljoscha Krettek
>Priority: Critical
>
> *Context*
> I run my [test
> code|https://gist.github.com/pbartoszek/fe6b1d7501333a2a4b385a1424f6bd80]
> against different triggers and runners. My original problem was that when
> writing to a file sink files weren't always produced in a deterministic way.
> Please refer to this
> [BEAM-3151|https://issues.apache.org/jira/browse/BEAM-3151] When I started
> looking at WriteFiles class I noticed that file sink implementation includes
> some multiple GroupByKey transformations. Knowing that I wrote my test code
> that is using multiple GroupByKey transformations to conclude that this is a
> bit buggy(?) support of After(Synchronised)ProcessingTime triggers by
> GroupByKey that also influence the file sink. When I run my job using
> Dataflow runner I was getting expected output.
> *About test code*
> The job is counting how many A and B elements it received within 30 sec
> windows using Count.perElement. Then I am using GroupByKey to fire every time
> count has increased.
> Below I outlined the expected standard output:
> {code:java}
> Let's assume all events are received in the same 30 seconds window.
> A -> After count KV{A, 1} -> Final group by KV{A, [1]}
> A -> After count KV{A, 2} -> Final group by KV{A, [1,2]}
> A -> After count KV{A, 3} -> Final group by KV{A, [1,2,3]}
> B -> After count KV{B, 1} -> Final group by KV{B, [1]}
> With my trigger configuration I would expect that for every new element
> 'After count' is printed with new value followed by 'Final group by' with new
> counter. Final group by represents the history of counters then.{code}
>
> *Problem*
> 'Final group by' trigger doesn't always go off although trigger set up would
> suggest that. This behaviour is different for different runners and Beam
> versions.
>
> *My observations when using Pubsub*
> Trigger configuration
> {code:java}
> Window.into(FixedWindows.of(standardSeconds(30)))
>
> .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()))
>
> .withAllowedLateness(standardSeconds(5),
> Window.ClosingBehavior.FIRE_IF_NON_EMPTY)
> .accumulatingFiredPanes())
> {code}
>
> Beam 2.0 Flink Runner
> {code:java}
> 2017-11-16T14:51:44.294Z After count KV{A, 1}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.036Z Received Element A
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.143Z After count KV{A, 2}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.246Z Final group by KV{A, [1, 2]}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:52:03.522Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:03.629Z After count KV{A, 1}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:03.732Z Final group by KV{A, [1]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.270Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.372Z After count KV{A, 2}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.476Z Final group by KV{A, [1, 2]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.394Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.501Z After count KV{A, 3}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.602Z Final group by KV{A, [1, 2, 3]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:13.296Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:13.402Z After count KV{A, 4}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:13.507Z Final group by KV{A, [1, 2, 3, 4]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:14.951Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z) <--- Expected to see
> 'After
[jira] [Commented] (BEAM-3225) Non deterministic behaviour of AfterProcessingTime trigger with multiple group by transformations
[
https://issues.apache.org/jira/browse/BEAM-3225?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16401800#comment-16401800
]
Dawid Wysakowicz commented on BEAM-3225:
Hi, I've tried to reproduce the _"allowed lateness configuration dictates that
only non empty panes should be trigger!!!"_ but couldn't do it by any means.
Also could not find any bug or issue in flink runner that could be reason for
such behaviour. [~pawelbartoszek] are you able to provide some reliable way to
reproduce the problem?
> Non deterministic behaviour of AfterProcessingTime trigger with multiple
> group by transformations
> -
>
> Key: BEAM-3225
> URL: https://issues.apache.org/jira/browse/BEAM-3225
> Project: Beam
> Issue Type: Bug
> Components: runner-core, runner-flink
>Affects Versions: 2.0.0, 2.1.0
>Reporter: Pawel Bartoszek
>Assignee: Aljoscha Krettek
>Priority: Critical
>
> *Context*
> I run my [test
> code|https://gist.github.com/pbartoszek/fe6b1d7501333a2a4b385a1424f6bd80]
> against different triggers and runners. My original problem was that when
> writing to a file sink files weren't always produced in a deterministic way.
> Please refer to this
> [BEAM-3151|https://issues.apache.org/jira/browse/BEAM-3151] When I started
> looking at WriteFiles class I noticed that file sink implementation includes
> some multiple GroupByKey transformations. Knowing that I wrote my test code
> that is using multiple GroupByKey transformations to conclude that this is a
> bit buggy(?) support of After(Synchronised)ProcessingTime triggers by
> GroupByKey that also influence the file sink. When I run my job using
> Dataflow runner I was getting expected output.
> *About test code*
> The job is counting how many A and B elements it received within 30 sec
> windows using Count.perElement. Then I am using GroupByKey to fire every time
> count has increased.
> Below I outlined the expected standard output:
> {code:java}
> Let's assume all events are received in the same 30 seconds window.
> A -> After count KV{A, 1} -> Final group by KV{A, [1]}
> A -> After count KV{A, 2} -> Final group by KV{A, [1,2]}
> A -> After count KV{A, 3} -> Final group by KV{A, [1,2,3]}
> B -> After count KV{B, 1} -> Final group by KV{B, [1]}
> With my trigger configuration I would expect that for every new element
> 'After count' is printed with new value followed by 'Final group by' with new
> counter. Final group by represents the history of counters then.{code}
>
> *Problem*
> 'Final group by' trigger doesn't always go off although trigger set up would
> suggest that. This behaviour is different for different runners and Beam
> versions.
>
> *My observations when using Pubsub*
> Trigger configuration
> {code:java}
> Window.into(FixedWindows.of(standardSeconds(30)))
>
> .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()))
>
> .withAllowedLateness(standardSeconds(5),
> Window.ClosingBehavior.FIRE_IF_NON_EMPTY)
> .accumulatingFiredPanes())
> {code}
>
> Beam 2.0 Flink Runner
> {code:java}
> 2017-11-16T14:51:44.294Z After count KV{A, 1}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.036Z Received Element A
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.143Z After count KV{A, 2}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.246Z Final group by KV{A, [1, 2]}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:52:03.522Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:03.629Z After count KV{A, 1}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:03.732Z Final group by KV{A, [1]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.270Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.372Z After count KV{A, 2}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.476Z Final group by KV{A, [1, 2]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.394Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.501Z After count KV{A, 3}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.602Z Final group by KV{A, [1, 2, 3]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:13.296Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:13.402Z After count KV{A, 4}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
>
[jira] [Commented] (BEAM-3225) Non deterministic behaviour of AfterProcessingTime trigger with multiple group by transformations
[
https://issues.apache.org/jira/browse/BEAM-3225?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16349669#comment-16349669
]
Eugene Kirpichov commented on BEAM-3225:
Thanks for the thorough investigation!
Another comment here: triggers *are* non-deterministic, in the sense that they
only *unblock* output from being produced, but don't cause it to be immediately
produced. Of course, runners generally try to produce it quickly, but e.g. with
an AfterPane.elementCountAtLeast(1) a runner is definitely allowed to process
several elements and then emit a single pane containing all of them, not
necessarily firing 1 pane for element (hence "element count at least 1" rather
than "exactly 1"). With something like AfterProcessingTime it is still more
vague, as clocks are approximate - by the time the trigger fires, more elements
could have arrived, so we can emit all of them.
To do what you want (full exact history of count changes) you'll need to write
a stateful DoFn with a per-key counter and manually emit everything you want.
That is what Beam does under the hood anyway (triggers are implemented using
per-key state and timers), just not with the semantics you want.
That explains all of your examples up until "Why Final group by is triggered
only after allowed lateness at 11:00:35?". As for that one, I'm not sure I
understand what is the issue? After Count A fires after the first time A
arrives (causing Final group by to fire), and fires another time when the
window closes, i.e. after the allowed lateness (causing Final group by to fire
one more time) - seems as expected?
As for "allowed lateness configuration dictates that only non empty panes
should be trigger!!!" - yes, this seems like a bug in the Flink runner;
[~aljoscha] could you take a look?
> Non deterministic behaviour of AfterProcessingTime trigger with multiple
> group by transformations
> -
>
> Key: BEAM-3225
> URL: https://issues.apache.org/jira/browse/BEAM-3225
> Project: Beam
> Issue Type: Bug
> Components: runner-core, runner-flink
>Affects Versions: 2.0.0, 2.1.0
>Reporter: Pawel Bartoszek
>Assignee: Eugene Kirpichov
>Priority: Critical
>
> *Context*
> I run my [test
> code|https://gist.github.com/pbartoszek/fe6b1d7501333a2a4b385a1424f6bd80]
> against different triggers and runners. My original problem was that when
> writing to a file sink files weren't always produced in a deterministic way.
> Please refer to this
> [BEAM-3151|https://issues.apache.org/jira/browse/BEAM-3151] When I started
> looking at WriteFiles class I noticed that file sink implementation includes
> some multiple GroupByKey transformations. Knowing that I wrote my test code
> that is using multiple GroupByKey transformations to conclude that this is a
> bit buggy(?) support of After(Synchronised)ProcessingTime triggers by
> GroupByKey that also influence the file sink. When I run my job using
> Dataflow runner I was getting expected output.
> *About test code*
> The job is counting how many A and B elements it received within 30 sec
> windows using Count.perElement. Then I am using GroupByKey to fire every time
> count has increased.
> Below I outlined the expected standard output:
> {code:java}
> Let's assume all events are received in the same 30 seconds window.
> A -> After count KV{A, 1} -> Final group by KV{A, [1]}
> A -> After count KV{A, 2} -> Final group by KV{A, [1,2]}
> A -> After count KV{A, 3} -> Final group by KV{A, [1,2,3]}
> B -> After count KV{B, 1} -> Final group by KV{B, [1]}
> With my trigger configuration I would expect that for every new element
> 'After count' is printed with new value followed by 'Final group by' with new
> counter. Final group by represents the history of counters then.{code}
>
> *Problem*
> 'Final group by' trigger doesn't always go off although trigger set up would
> suggest that. This behaviour is different for different runners and Beam
> versions.
>
> *My observations when using Pubsub*
> Trigger configuration
> {code:java}
> Window.into(FixedWindows.of(standardSeconds(30)))
>
> .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()))
>
> .withAllowedLateness(standardSeconds(5),
> Window.ClosingBehavior.FIRE_IF_NON_EMPTY)
> .accumulatingFiredPanes())
> {code}
>
> Beam 2.0 Flink Runner
> {code:java}
> 2017-11-16T14:51:44.294Z After count KV{A, 1}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.036Z Received Element A
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.143Z After count KV{A, 2}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
>
[jira] [Commented] (BEAM-3225) Non deterministic behaviour of AfterProcessingTime trigger with multiple group by transformations
[
https://issues.apache.org/jira/browse/BEAM-3225?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16332980#comment-16332980
]
Kenneth Knowles commented on BEAM-3225:
---
Eugene has the most context on this, as he recently really dug into the issue
of continuation triggers, triggers that terminate, and file sinks.
> Non deterministic behaviour of AfterProcessingTime trigger with multiple
> group by transformations
> -
>
> Key: BEAM-3225
> URL: https://issues.apache.org/jira/browse/BEAM-3225
> Project: Beam
> Issue Type: Bug
> Components: runner-core, runner-flink
>Affects Versions: 2.0.0, 2.1.0
>Reporter: Pawel Bartoszek
>Assignee: Eugene Kirpichov
>Priority: Critical
>
> *Context*
> I run my [test
> code|https://gist.github.com/pbartoszek/fe6b1d7501333a2a4b385a1424f6bd80]
> against different triggers and runners. My original problem was that when
> writing to a file sink files weren't always produced in a deterministic way.
> Please refer to this
> [BEAM-3151|https://issues.apache.org/jira/browse/BEAM-3151] When I started
> looking at WriteFiles class I noticed that file sink implementation includes
> some multiple GroupByKey transformations. Knowing that I wrote my test code
> that is using multiple GroupByKey transformations to conclude that this is a
> bit buggy(?) support of After(Synchronised)ProcessingTime triggers by
> GroupByKey that also influence the file sink. When I run my job using
> Dataflow runner I was getting expected output.
> *About test code*
> The job is counting how many A and B elements it received within 30 sec
> windows using Count.perElement. Then I am using GroupByKey to fire every time
> count has increased.
> Below I outlined the expected standard output:
> {code:java}
> Let's assume all events are received in the same 30 seconds window.
> A -> After count KV{A, 1} -> Final group by KV{A, [1]}
> A -> After count KV{A, 2} -> Final group by KV{A, [1,2]}
> A -> After count KV{A, 3} -> Final group by KV{A, [1,2,3]}
> B -> After count KV{B, 1} -> Final group by KV{B, [1]}
> With my trigger configuration I would expect that for every new element
> 'After count' is printed with new value followed by 'Final group by' with new
> counter. Final group by represents the history of counters then.{code}
>
> *Problem*
> 'Final group by' trigger doesn't always go off although trigger set up would
> suggest that. This behaviour is different for different runners and Beam
> versions.
>
> *My observations when using Pubsub*
> Trigger configuration
> {code:java}
> Window.into(FixedWindows.of(standardSeconds(30)))
>
> .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()))
>
> .withAllowedLateness(standardSeconds(5),
> Window.ClosingBehavior.FIRE_IF_NON_EMPTY)
> .accumulatingFiredPanes())
> {code}
>
> Beam 2.0 Flink Runner
> {code:java}
> 2017-11-16T14:51:44.294Z After count KV{A, 1}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.036Z Received Element A
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.143Z After count KV{A, 2}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.246Z Final group by KV{A, [1, 2]}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:52:03.522Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:03.629Z After count KV{A, 1}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:03.732Z Final group by KV{A, [1]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.270Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.372Z After count KV{A, 2}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.476Z Final group by KV{A, [1, 2]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.394Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.501Z After count KV{A, 3}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.602Z Final group by KV{A, [1, 2, 3]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:13.296Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:13.402Z After count KV{A, 4}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:13.507Z Final group by KV{A, [1, 2, 3, 4]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:14.951Z Received Element A
>
[jira] [Commented] (BEAM-3225) Non deterministic behaviour of AfterProcessingTime trigger with multiple group by transformations
[
https://issues.apache.org/jira/browse/BEAM-3225?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16332979#comment-16332979
]
Kenneth Knowles commented on BEAM-3225:
---
I haven't read the whole trace you posted yet, but I want to highlight this:
Only the DataflowRunner and DirectRunner support the "synchronized" part of
synchronized processing time. And we basically intend to drop it. We've had
some discussions about dangerous behavior that it causes. Essentially, if you
choose a repeating processing time trigger, you should expect all your data out
in some number of panes from the second GBK, but you can't be too picky about
how many.
> Non deterministic behaviour of AfterProcessingTime trigger with multiple
> group by transformations
> -
>
> Key: BEAM-3225
> URL: https://issues.apache.org/jira/browse/BEAM-3225
> Project: Beam
> Issue Type: Bug
> Components: runner-core, runner-flink
>Affects Versions: 2.0.0, 2.1.0
>Reporter: Pawel Bartoszek
>Assignee: Kenneth Knowles
>Priority: Critical
>
> *Context*
> I run my [test
> code|https://gist.github.com/pbartoszek/fe6b1d7501333a2a4b385a1424f6bd80]
> against different triggers and runners. My original problem was that when
> writing to a file sink files weren't always produced in a deterministic way.
> Please refer to this
> [BEAM-3151|https://issues.apache.org/jira/browse/BEAM-3151] When I started
> looking at WriteFiles class I noticed that file sink implementation includes
> some multiple GroupByKey transformations. Knowing that I wrote my test code
> that is using multiple GroupByKey transformations to conclude that this is a
> bit buggy(?) support of After(Synchronised)ProcessingTime triggers by
> GroupByKey that also influence the file sink. When I run my job using
> Dataflow runner I was getting expected output.
> *About test code*
> The job is counting how many A and B elements it received within 30 sec
> windows using Count.perElement. Then I am using GroupByKey to fire every time
> count has increased.
> Below I outlined the expected standard output:
> {code:java}
> Let's assume all events are received in the same 30 seconds window.
> A -> After count KV{A, 1} -> Final group by KV{A, [1]}
> A -> After count KV{A, 2} -> Final group by KV{A, [1,2]}
> A -> After count KV{A, 3} -> Final group by KV{A, [1,2,3]}
> B -> After count KV{B, 1} -> Final group by KV{B, [1]}
> With my trigger configuration I would expect that for every new element
> 'After count' is printed with new value followed by 'Final group by' with new
> counter. Final group by represents the history of counters then.{code}
>
> *Problem*
> 'Final group by' trigger doesn't always go off although trigger set up would
> suggest that. This behaviour is different for different runners and Beam
> versions.
>
> *My observations when using Pubsub*
> Trigger configuration
> {code:java}
> Window.into(FixedWindows.of(standardSeconds(30)))
>
> .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()))
>
> .withAllowedLateness(standardSeconds(5),
> Window.ClosingBehavior.FIRE_IF_NON_EMPTY)
> .accumulatingFiredPanes())
> {code}
>
> Beam 2.0 Flink Runner
> {code:java}
> 2017-11-16T14:51:44.294Z After count KV{A, 1}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.036Z Received Element A
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.143Z After count KV{A, 2}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:51:53.246Z Final group by KV{A, [1, 2]}
> [2017-11-16T14:51:30.000Z..2017-11-16T14:52:00.000Z)
> 2017-11-16T14:52:03.522Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:03.629Z After count KV{A, 1}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:03.732Z Final group by KV{A, [1]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.270Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.372Z After count KV{A, 2}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:07.476Z Final group by KV{A, [1, 2]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.394Z Received Element A
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.501Z After count KV{A, 3}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:10.602Z Final group by KV{A, [1, 2, 3]}
> [2017-11-16T14:52:00.000Z..2017-11-16T14:52:30.000Z)
> 2017-11-16T14:52:13.296Z Received Element A
>
