[beam] branch master updated: [BEAM-12842] Add timestamp to test work item to deflake

[ningk]

This is an automated email from the ASF dual-hosted git repository.

ningk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 0d02a88  [BEAM-12842] Add timestamp to test work item to deflake
     new 1547518  Merge pull request #15492 from rohdesamuel/hotkeyflake
0d02a88 is described below

commit 0d02a8801507fd6e45592ab78b36109db8006de0
Author: Sam R <rohde.sam...@gmail.com>
AuthorDate: Fri Sep 10 08:33:22 2021 -0700

    [BEAM-12842] Add timestamp to test work item to deflake
---
 .../beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java   | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git 
a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java
 
b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java
index 12b7df2..17b59ec 100644
--- 
a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java
+++ 
b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java
@@ -794,7 +794,8 @@ public class StreamingDataflowWorkerTest {
 
     final int numIters = 2000;
     for (int i = 0; i < numIters; ++i) {
-      server.addWorkToOffer(makeInput(i, 0, "key", DEFAULT_SHARDING_KEY));
+      server.addWorkToOffer(
+          makeInput(i, TimeUnit.MILLISECONDS.toMicros(i), "key", 
DEFAULT_SHARDING_KEY));
     }
 
     Map<Long, Windmill.WorkItemCommitRequest> result = 
server.waitForAndGetCommits(numIters);
@@ -829,7 +830,8 @@ public class StreamingDataflowWorkerTest {
 
     final int numIters = 2000;
     for (int i = 0; i < numIters; ++i) {
-      server.addWorkToOffer(makeInput(i, 0, "key", DEFAULT_SHARDING_KEY));
+      server.addWorkToOffer(
+          makeInput(i, TimeUnit.MILLISECONDS.toMicros(i), "key", 
DEFAULT_SHARDING_KEY));
     }
 
     Map<Long, Windmill.WorkItemCommitRequest> result = 
server.waitForAndGetCommits(numIters);