fix: attempt to fix tests

Author: François Delbrayelle

plugin-debezium-postgres/src/test/java/io/kestra/plugin/debezium/postgres/CaptureTest.java

@@ -63,6 +63,7 @@ void run() throws Exception {
             .port(Property.ofValue("65432"))
             .database(Property.ofValue("postgres"))
             .pluginName(Property.ofValue(PostgresInterface.PluginName.PGOUTPUT))
+            .stateName(Property.ofValue("debezium-state-" + IdUtils.create()))
             // SSL is disabled or we cannot test triggers which are very important for Debezium
 //            .sslMode(TestUtils.sslMode())
 //            .sslRootCert(TestUtils.ca())

plugin-debezium-postgres/src/test/java/io/kestra/plugin/debezium/postgres/PostgresDebeziumTestHelper.java

@@ -12,6 +12,7 @@
 import io.kestra.plugin.debezium.AbstractDebeziumTask;
 
 import java.sql.SQLException;
+import java.util.Arrays;
 import java.util.concurrent.Callable;
 
 final class PostgresDebeziumTestHelper {
@@ -40,14 +41,14 @@ static void dropReplicationArtifacts(Callable<java.sql.Connection> connectionSup
     }
 
     static void cleanupTaskState(RunContext runContext, AbstractDebeziumTask task) throws Exception {
-        KVStore kvStore = runContext.namespaceKv(runContext.flowInfo().namespace());
-        String taskRunValue = runContext.storage().getTaskStorageContext()
+        var kvStore = runContext.namespaceKv(runContext.flowInfo().namespace());
+        var taskRunValue = runContext.storage().getTaskStorageContext()
             .map(StorageContext.Task::getTaskRunValue)
             .orElse(null);
-        String stateName = runContext.render(task.getStateName()).as(String.class).orElse("debezium-state");
+        var stateName = runContext.render(task.getStateName()).as(String.class).orElse("debezium-state");
 
-        kvStore.delete(computeKvStoreKey(runContext, stateName, "offsets.dat", taskRunValue));
-        kvStore.delete(computeKvStoreKey(runContext, stateName, "dbhistory.dat", taskRunValue));
+        deleteAllVersions(kvStore, computeKvStoreKey(runContext, stateName, "offsets.dat", taskRunValue));
+        deleteAllVersions(kvStore, computeKvStoreKey(runContext, stateName, "dbhistory.dat", taskRunValue));
     }
 
     static void cleanupFlowState(KVStoreService kvStoreService, String namespace, String flowId, String... stateNames) throws Exception {
@@ -58,14 +59,15 @@ static void cleanupFlowState(KVStoreService kvStoreService, String namespace, St
             return;
         }
 
-        String flowPrefix = Slugify.of(flowId) + "_states_";
+        var flowPrefix = Slugify.of(flowId) + "_states_";
+        var keysToDelete = kvStore.listAll().stream()
+            .map(KVEntry::key)
+            .distinct()
+            .filter(key -> Arrays.stream(stateNames).anyMatch(stateName -> key.startsWith(flowPrefix + stateName)))
+            .toList();
 
-        for (KVEntry kvEntry : kvStore.listAll()) {
-            for (String stateName : stateNames) {
-                if (kvEntry.key().startsWith(flowPrefix + stateName)) {
-                    kvStore.delete(kvEntry.key());
-                }
-            }
+        for (var key : keysToDelete) {
+            deleteAllVersions(kvStore, key);
         }
     }
 
@@ -84,4 +86,10 @@ private static String computeKvStoreKey(RunContext runContext, String stateName,
 
         return prefix + separator + filename;
     }
+
+    private static void deleteAllVersions(KVStore kvStore, String key) throws Exception {
+        while (kvStore.delete(key)) {
+            // Delete all versions to prevent fallback to stale previous entries.
+        }
+    }
 }

plugin-debezium-postgres/src/test/resources/flows/realtime.yaml

@@ -9,6 +9,7 @@ triggers:
     port: 65432
     username: postgres
     password: pg_passwd
+    stateName: "debezium-state-{{ now() | date('yyyyMMddHHmmssSSS') }}"
     includedTables:
       - public.events
 

plugin-debezium-postgres/src/test/resources/flows/trigger.yaml

@@ -9,6 +9,7 @@ triggers:
     port: 65432
     username: postgres
     password: pg_passwd
+    stateName: "debezium-state-{{ now() | date('yyyyMMddHHmmssSSS') }}"
     maxRecords: 5
     includedTables:
       - public.events