Fix StatefulStreamingParDoEvaluatorTest window assertions

The createStreamingSource helper method generates events across 4 different
1-second windows when iterCount=2, because it advances the watermark after
every single batch (even between Key 1 and Key 2 batches).

Previously, VerifyStatefulOutput incorrectly assumed Key 1 and Key 2 would
be emitted in the same windows (Window 1 and Window 2). This caused an
AssertionError because Key 2 actually had 0 elements in Window 1.

This commit updates the assertions in shouldProcessWindowedStatefulParDo
to verify the 4 distinct windows individually, matching the generated
timestamps.

Author: durgaprasadml

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

@@ -372,14 +372,24 @@ public void shouldProcessWindowedStatefulParDo() {
 
     IntervalWindow window1 = new IntervalWindow(new Instant(0), Duration.standardSeconds(1L));
     IntervalWindow window2 = new IntervalWindow(new Instant(1000), Duration.standardSeconds(1L));
+    IntervalWindow window3 = new IntervalWindow(new Instant(2000), Duration.standardSeconds(1L));
+    IntervalWindow window4 = new IntervalWindow(new Instant(3000), Duration.standardSeconds(1L));
 
     PAssert.that(result)
         .inWindow(window1)
-        .satisfies(new VerifyStatefulOutput(Arrays.asList(1, 2, 3), Arrays.asList(4, 5, 6)));
+        .satisfies(new VerifyStatefulOutput(Arrays.asList(1, 2, 3), Collections.emptyList()));
 
     PAssert.that(result)
         .inWindow(window2)
-        .satisfies(new VerifyStatefulOutput(Arrays.asList(7, 8, 9), Arrays.asList(10, 11, 12)));
+        .satisfies(new VerifyStatefulOutput(Collections.emptyList(), Arrays.asList(4, 5, 6)));
+
+    PAssert.that(result)
+        .inWindow(window3)
+        .satisfies(new VerifyStatefulOutput(Arrays.asList(7, 8, 9), Collections.emptyList()));
+
+    PAssert.that(result)
+        .inWindow(window4)
+        .satisfies(new VerifyStatefulOutput(Collections.emptyList(), Arrays.asList(10, 11, 12)));
 
     p.run().waitUntilFinish();
   }