[flink][test] fix Flink 2.x open() compatibility and flaky snapshot trigger

- ScanAndCleanFunction: change open(Configuration) to open(OpenContext)
  for Flink 2.x compatibility (Flink 2.x removed the Configuration overload)
- FlussClusterExtension: triggerSnapshot() returns null on no-op instead of
  failing hard when snapshot ID does not advance (initSnapshot skips when
  logOffset <= lastSnapshotOffset)
- triggerAndWaitSnapshots() silently skips null buckets (original behavior)

Author: platinumhamburg

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java

@@ -77,7 +77,9 @@ public ScanAndCleanFunction(long deleteRateLimitPerSecond, Map<String, String> e
     }
 
     @Override
-    public void open(org.apache.flink.configuration.Configuration parameters) {
+    public void open(org.apache.flink.api.common.functions.OpenContext openContext)
+            throws Exception {
+        super.open(openContext);
         if (!extraConfigs.isEmpty()) {
             FileSystem.initialize(Configuration.fromMap(extraConfigs), null);
         }

fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java

@@ -771,33 +771,33 @@ public CompletedSnapshot triggerAndWaitSnapshot(TableBucket tableBucket) {
     }
 
     private Long triggerSnapshot(TableBucket tableBucket) {
-        Long snapshotId = null;
-        Long nextSnapshotId = null;
         for (TabletServer ts : tabletServers.values()) {
             ReplicaManager.HostedReplica replica = ts.getReplicaManager().getReplica(tableBucket);
             if (replica instanceof ReplicaManager.OnlineReplica) {
                 Replica r = ((ReplicaManager.OnlineReplica) replica).getReplica();
                 PeriodicSnapshotManager kvSnapshotManager = r.getKvSnapshotManager();
                 if (r.isLeader() && kvSnapshotManager != null) {
-                    snapshotId = kvSnapshotManager.currentSnapshotId();
+                    long snapshotId = kvSnapshotManager.currentSnapshotId();
                     kvSnapshotManager.triggerSnapshot();
-                    nextSnapshotId = kvSnapshotManager.currentSnapshotId();
-                    break;
+                    // triggerSnapshot() submits to guardedExecutor asynchronously.
+                    // Wait for the counter to advance; if it does not, initSnapshot()
+                    // determined there is no new data (logOffset <= lastSnapshotOffset)
+                    // and the trigger was a legitimate no-op — return null.
+                    try {
+                        waitUntil(
+                                () -> kvSnapshotManager.currentSnapshotId() > snapshotId,
+                                Duration.ofSeconds(3),
+                                Duration.ofMillis(50),
+                                "");
+                    } catch (AssertionError e) {
+                        return null;
+                    }
+                    return snapshotId;
                 }
             }
         }
-
-        if (snapshotId != null) {
-            if (nextSnapshotId > snapshotId) {
-                // only there is a new snapshot triggered, we return the snapshot id
-                return snapshotId;
-            } else {
-                return null;
-            }
-        } else {
-            fail("No KV snapshot manager found for table bucket " + tableBucket);
-            return null;
-        }
+        fail("No KV snapshot manager found for table bucket " + tableBucket);
+        return null;
     }
 
     public CompletedSnapshot waitUntilSnapshotFinished(TableBucket tableBucket, long snapshotId) {