Repository: incubator-beam Updated Branches: refs/heads/master 9e0d7d650 -> e1b305ea5
Make example AddTimestampFn range deterministic The timestamps added in the WindowedWordCount example are currently based on when the bundles are executed, which makes the min/max bounds non-deterministic. This change makes the range based on the construction time. Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/335202a0 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/335202a0 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/335202a0 Branch: refs/heads/master Commit: 335202a033ced6f30f1b0e5df9da047241abc750 Parents: 9e0d7d6 Author: Scott Wegner <[email protected]> Authored: Thu Jun 9 11:31:23 2016 -0700 Committer: bchambers <[email protected]> Committed: Mon Jun 13 11:21:25 2016 -0700 ---------------------------------------------------------------------- .../java/org/apache/beam/examples/WindowedWordCount.java | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/335202a0/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java b/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java index 4d019bb..9ca26bf 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java +++ b/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java @@ -124,13 +124,18 @@ public class WindowedWordCount { * 2-hour period. */ static class AddTimestampFn extends DoFn<String, String> { - private static final long RAND_RANGE = 7200000; // 2 hours in ms + private static final Duration RAND_RANGE = Duration.standardHours(2); + private final Instant minTimestamp; + + AddTimestampFn() { + this.minTimestamp = new Instant(System.currentTimeMillis()); + } @Override public void processElement(ProcessContext c) { // Generate a timestamp that falls somewhere in the past two hours. - long randomTimestamp = System.currentTimeMillis() - - (int) (Math.random() * RAND_RANGE); + long randMillis = (long) (Math.random() * RAND_RANGE.getMillis()); + Instant randomTimestamp = minTimestamp.plus(randMillis); /** * Concept #2: Set the data element with that timestamp. */
