Fix Flaky PreCommit Spark runner Java version test (#34974)

Author: Abacn

runners/spark/3/build.gradle

@@ -34,17 +34,11 @@ createJavaExamplesArchetypeValidationTask(type: 'Quickstart', runner: 'Spark')
 
 // Additional supported Spark versions (used in compatibility tests)
 def sparkVersions = [
-    "350": "3.5.0",
-    "341": "3.4.1",
-    "340": "3.4.0",
-    "332": "3.3.2",
-    "331": "3.3.1",
-    "330": "3.3.0",
+    "350": "3.5.5",
+    "341": "3.4.4",
+    "332": "3.3.4",
     "324": "3.2.4",
-    "323": "3.2.3",
-    "321": "3.2.1",
-    "312": "3.1.2",
-    "311": "3.1.1"
+    "312": "3.1.3",
 ]
 
 sparkVersions.each { kv ->
@@ -94,3 +88,7 @@ tasks.register("sparkVersionsTest") {
   group = "Verification"
   dependsOn sparkVersions.collect{k,v -> "sparkVersion${k}Test"}
 }
+
+tasks.test {
+  outputs.upToDateWhen { false }
+}

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

@@ -19,9 +19,10 @@
 
 import static org.apache.beam.sdk.metrics.MetricResultsMatchers.attemptedMetricsResult;
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
 import java.io.Serializable;
@@ -250,15 +251,15 @@ public void testFlattenPCollResumeFromCheckpoint() {
                 "BoundedAssert",
                 DistributionResult.create(45, 10, 0L, 9L))));
 
-    // Verify metrics for Flattened result after first run
-    assertThat(
-        res.metrics().queryMetrics(metricsFilter).getDistributions(),
-        hasItem(
-            attemptedMetricsResult(
-                PAssertFn.class.getName(),
-                "distribution",
-                "FlattenedAssert",
-                DistributionResult.create(45, 10, 0L, 9L))));
+    // Fetch metrics for Flattened result after first run
+    long firstMax = 0;
+    for (MetricResult<DistributionResult> dists :
+        res.metrics().queryMetrics(metricsFilter).getDistributions()) {
+      long currentMax = dists.getAttempted().getMax();
+      if (currentMax > firstMax) {
+        firstMax = currentMax;
+      }
+    }
 
     // Clean up state
     clean();
@@ -276,29 +277,19 @@ public void testFlattenPCollResumeFromCheckpoint() {
                 "BoundedAssert",
                 DistributionResult.create(45, 10, 0L, 9L))));
 
-    // Verify Flattened results show accumulated values from both runs
-    // We use anyOf matcher because the unbounded source may emit either 2 or 3 elements during the
-    // test window:
-    // Case 1 (3 elements): sum=78 (45 from bounded + 33 from unbounded), count=13 (10 bounded + 3
-    // unbounded)
-    // Case 2 (2 elements): sum=66 (45 from bounded + 21 from unbounded), count=12 (10 bounded + 2
-    // unbounded)
-    // This variation occurs because the unbounded source's withRate(3, Duration.standardSeconds(1))
-    // timing may be affected by test environment conditions
-    assertThat(
-        res.metrics().queryMetrics(metricsFilter).getDistributions(),
-        hasItem(
-            anyOf(
-                attemptedMetricsResult(
-                    PAssertFn.class.getName(),
-                    "distribution",
-                    "FlattenedAssert",
-                    DistributionResult.create(78, 13, 0, 12)),
-                attemptedMetricsResult(
-                    PAssertFn.class.getName(),
-                    "distribution",
-                    "FlattenedAssert",
-                    DistributionResult.create(66, 12, 0, 11)))));
+    long secondMax = 0;
+    long secondSum = 0;
+    for (MetricResult<DistributionResult> dists :
+        res.metrics().queryMetrics(metricsFilter).getDistributions()) {
+      long currentMax = dists.getAttempted().getMax();
+      if (currentMax > secondMax) {
+        secondMax = currentMax;
+        secondSum = dists.getAttempted().getSum();
+      }
+    }
+
+    assertTrue(secondMax > firstMax);
+    assertEquals((1L + secondMax) * secondMax / 2, secondSum);
   }
 
   /** Restarts the pipeline from checkpoint. Sets pipeline to stop after 1 second. */