Restore strong validation semantics in StatefulStreamingParDoEvaluatorTest

Author: durgaprasadml

runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StatefulStreamingParDoEvaluatorTest.java

@@ -75,15 +75,13 @@ private PTransform<PBegin, PCollection<KV<Integer, Integer>>> createStreamingSou
     return CreateStream.of(coder, batchDuration)
         .emptyBatch()
         .advanceWatermarkForNextBatch(instant)
-        .nextBatch(
-            TimestampedValue.of(KV.of(1, 10), instant),
-            TimestampedValue.of(KV.of(1, 10), instant),
-            TimestampedValue.of(KV.of(1, 10), instant))
+        .nextBatch(TimestampedValue.of(KV.of(1, 1), instant))
+        .nextBatch(TimestampedValue.of(KV.of(1, 2), instant))
+        .nextBatch(TimestampedValue.of(KV.of(1, 3), instant))
         .advanceWatermarkForNextBatch(instant.plus(Duration.standardSeconds(1L)))
-        .nextBatch(
-            TimestampedValue.of(KV.of(2, 20), instant.plus(Duration.standardSeconds(1L))),
-            TimestampedValue.of(KV.of(2, 20), instant.plus(Duration.standardSeconds(1L))),
-            TimestampedValue.of(KV.of(2, 20), instant.plus(Duration.standardSeconds(1L))))
+        .nextBatch(TimestampedValue.of(KV.of(2, 4), instant.plus(Duration.standardSeconds(1L))))
+        .nextBatch(TimestampedValue.of(KV.of(2, 5), instant.plus(Duration.standardSeconds(1L))))
+        .nextBatch(TimestampedValue.of(KV.of(2, 6), instant.plus(Duration.standardSeconds(1L))))
         .advanceNextBatchWatermarkToInfinity();
   }
 
@@ -96,21 +94,22 @@ private PTransform<PBegin, PCollection<KV<Integer, Integer>>> createStreamingSou
     CreateStream<KV<Integer, Integer>> createStream =
         CreateStream.of(coder, batchDuration).emptyBatch().advanceWatermarkForNextBatch(instant);
 
+    int value = 1;
     for (int i = 0; i < iterCount; i++) {
       createStream =
-          createStream.nextBatch(
-              TimestampedValue.of(KV.of(1, 10), instant),
-              TimestampedValue.of(KV.of(1, 10), instant),
-              TimestampedValue.of(KV.of(1, 10), instant));
+          createStream
+              .nextBatch(TimestampedValue.of(KV.of(1, value++), instant))
+              .nextBatch(TimestampedValue.of(KV.of(1, value++), instant))
+              .nextBatch(TimestampedValue.of(KV.of(1, value++), instant));
 
       instant = instant.plus(Duration.standardSeconds(1L));
       createStream = createStream.advanceWatermarkForNextBatch(instant);
 
       createStream =
-          createStream.nextBatch(
-              TimestampedValue.of(KV.of(2, 20), instant),
-              TimestampedValue.of(KV.of(2, 20), instant),
-              TimestampedValue.of(KV.of(2, 20), instant));
+          createStream
+              .nextBatch(TimestampedValue.of(KV.of(2, value++), instant))
+              .nextBatch(TimestampedValue.of(KV.of(2, value++), instant))
+              .nextBatch(TimestampedValue.of(KV.of(2, value++), instant));
 
       instant = instant.plus(Duration.standardSeconds(1L));
       createStream = createStream.advanceWatermarkForNextBatch(instant);
@@ -310,13 +309,13 @@ public void shouldProcessGlobalWidowStatefulParDo() {
     PAssert.that(result)
         .containsInAnyOrder(
             // key 1
-            KV.of(1, 10), // 10
-            KV.of(1, 20), // 10 + 10
-            KV.of(1, 30), // 20 + 10
+            KV.of(1, 1), // 1
+            KV.of(1, 3), // 1 + 2
+            KV.of(1, 6), // 3 + 3
             // key 2
-            KV.of(2, 20), // 20
-            KV.of(2, 40), // 20 + 20
-            KV.of(2, 60)); // 40 + 20
+            KV.of(2, 4), // 4
+            KV.of(2, 9), // 4 + 5
+            KV.of(2, 15)); // 9 + 6
 
     p.run().waitUntilFinish();
   }
@@ -331,25 +330,25 @@ public void shouldProcessWindowedStatefulParDo() {
 
     PAssert.that(result)
         .containsInAnyOrder(
-            // Windowed Key 1 (Window 1)
-            KV.of(1, 10), // 10
-            KV.of(1, 20), // 10 + 10
-            KV.of(1, 30), // 20 + 10
-
-            // Windowed Key 2 (Window 1)
-            KV.of(2, 20), // 20
-            KV.of(2, 40), // 20 + 20
-            KV.of(2, 60), // 40 + 20
-
-            // Windowed Key 1 (Window 2)
-            KV.of(1, 10), // 10
-            KV.of(1, 20), // 10 + 10
-            KV.of(1, 30), // 20 + 10
-
-            // Windowed Key 2 (Window 2)
-            KV.of(2, 20), // 20
-            KV.of(2, 40), // 20 + 20
-            KV.of(2, 60) // 40 + 20
+            // Windowed Key 1
+            KV.of(1, 1), // 1
+            KV.of(1, 3), // 1 + 2
+            KV.of(1, 6), // 3 + 3
+
+            // Windowed Key 2
+            KV.of(2, 4), // 4
+            KV.of(2, 9), // 4 + 5
+            KV.of(2, 15), // 9 + 6
+
+            // Windowed Key 1
+            KV.of(1, 7), // 7
+            KV.of(1, 15), // 7 + 8
+            KV.of(1, 24), // 15 + 9
+
+            // Windowed Key 2
+            KV.of(2, 10), // 10
+            KV.of(2, 21), // 10 + 11
+            KV.of(2, 33) // 21 + 12
             );
 
     p.run().waitUntilFinish();