[FLINK-11041][test] ReinterpretDataStreamAsKeyedStreamITCase source s…

…hould hold checkpointing lock
igalshilman · Dec 11, 2018 · 70b2029 · 70b2029
1 parent 927936d
commit 70b2029
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/...a/org/apache/flink/streaming/api/datastream/ReinterpretDataStreamAsKeyedStreamITCase.java b/...a/org/apache/flink/streaming/api/datastream/ReinterpretDataStreamAsKeyedStreamITCase.java
@@ -122,7 +122,9 @@ public RandomTupleSource(int numEvents, int numKeys) {
 		public void run(SourceContext<Tuple2<Integer, Integer>> out) throws Exception {
 			Random random = new Random(42);
 			while (--remainingEvents >= 0) {
-				out.collect(new Tuple2<>(random.nextInt(numKeys), 1));
+				synchronized (out.getCheckpointLock()) {
+					out.collect(new Tuple2<>(random.nextInt(numKeys), 1));
+				}
 			}
 		}