[test][pipeline-e2e] Tolerate transient regressed reads in MySqlToHudiE2eITCase

Hudi MERGE_ON_READ snapshot reads can momentarily expose an empty or
partial file slice during compaction, making validateSinkResult judge a
transient empty result and report a misleading Actual:[]. Keep the best
observed read and skip regressed reads so the final assertion never lands
on a transient empty slice.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: leonardBang

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToHudiE2eITCase.java

@@ -491,9 +491,30 @@ private void validateSinkResult(
         LOG.info("Verifying Hudi {}::{}::{} results...", warehouse, database, table);
         long deadline = System.currentTimeMillis() + HUDI_TESTCASE_TIMEOUT.toMillis();
         List<String> results = Collections.emptyList();
+        int maxObservedSize = -1;
         while (System.currentTimeMillis() < deadline) {
             try {
-                results = fetchHudiTableRows(warehouse, database, table);
+                List<String> fetched = fetchHudiTableRows(warehouse, database, table);
+
+                // Hudi MERGE_ON_READ tables can momentarily expose an empty or partial file
+                // slice while a compaction swaps slices, so a snapshot read may regress to fewer
+                // rows (sometimes 0) even though no data was actually lost. The row count in this
+                // test only grows, so treat a regressed read as a transient inconsistent state:
+                // skip judging it this round and re-read on the next loop, keeping the best
+                // (largest) read so the final assertion never lands on a transient empty result.
+                if (maxObservedSize > 0 && fetched.size() < maxObservedSize) {
+                    LOG.warn(
+                            "Ignoring transient regressed read from Hudi MOR table: got {} rows, "
+                                    + "previously saw {} rows (likely a compaction file-slice swap). "
+                                    + "Waiting for the next loop...",
+                            fetched.size(),
+                            maxObservedSize);
+                    Thread.sleep(10000L);
+                    continue;
+                }
+
+                results = fetched;
+                maxObservedSize = fetched.size();
                 Assertions.assertThat(results).containsExactlyInAnyOrderElementsOf(expected);
                 LOG.info(
                         "Successfully verified {} records in {} seconds for {}::{}.",