Wait for parked backup execution in vector optimization tests [AI-192]

java/drivers/driver-hazelcast4plus/src/main/java/com/hazelcast/simulator/tests/vector/VectorCollectionPutDatasetTest.java

@@ -78,15 +78,16 @@ public void delete() {
     public void optimize() throws ExecutionException, InterruptedException {
         var start = System.currentTimeMillis();
 
-        // TODO: this may fail with > 1 client
+        // TODO: this will cause redundant invocations with >1 client
         var cleanupTimer = withTimer(() -> collection.optimizeAsync().toCompletableFuture().join());
 
         if (backupCount + asyncBackupCount > 0) {
-            // optimize backups are async - wait for them also to get true time
+            // optimize backups are async and blocking (can be parked) - wait for them also to get true time
             HazelcastUtils.waitForClusterSafeState(targetInstance);
+            HazelcastUtils.waitForNoParkedOperations(targetInstance);
             var cleanupTotalTimer = System.currentTimeMillis() - start;
             logger.info("Cleanup primary time: {} ms", cleanupTimer);
-            logger.info("Cleanup wait for backups time: {} ms", cleanupTotalTimer -cleanupTimer);
+            logger.info("Cleanup wait for backups time: {} ms", cleanupTotalTimer - cleanupTimer);
             logger.info("Cleanup total time: {} ms", cleanupTotalTimer);
         } else {
             logger.info("Cleanup time: {} ms", cleanupTimer);

java/drivers/driver-hazelcast4plus/src/main/java/com/hazelcast/simulator/utils/HazelcastUtils.java

@@ -19,8 +19,10 @@
 import com.hazelcast.config.Config;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.HazelcastInstanceAware;
+import com.hazelcast.instance.impl.HazelcastInstanceImpl;
 import com.hazelcast.shaded.org.json.JSONArray;
 import com.hazelcast.shaded.org.json.JSONObject;
+import com.hazelcast.spi.impl.operationparker.impl.OperationParkerImpl;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -128,6 +130,13 @@ public static void waitForClusterSafeState(HazelcastInstance targetInstance) thr
         targetInstance.getExecutorService("safe-check").submit(new WaitForClusterSafe()).get();
     }
 
+    public static void waitForNoParkedOperations(HazelcastInstance targetInstance) throws InterruptedException, ExecutionException {
+        var futures = targetInstance.getExecutorService("safe-check").submitToAllMembers(new WaitForNoParkedOperations());
+        for (var future : futures.values()) {
+            future.get();
+        }
+    }
+
     private static class WaitForClusterSafe implements Callable<Boolean>, HazelcastInstanceAware, Serializable {
         private HazelcastInstance node;
 
@@ -144,4 +153,22 @@ public void setHazelcastInstance(HazelcastInstance node) {
             this.node = node;
         }
     }
+
+    private static class WaitForNoParkedOperations implements Callable<Boolean>, HazelcastInstanceAware, Serializable {
+        private HazelcastInstance node;
+
+        @Override
+        public Boolean call() throws Exception {
+            OperationParkerImpl parker = (OperationParkerImpl) ((HazelcastInstanceImpl) node).node.getNodeEngine().getOperationParker();
+            while (parker.getTotalValidWaitingOperationCount() > 0) {
+                Thread.sleep(1);
+            }
+            return true;
+        }
+
+        @Override
+        public void setHazelcastInstance(HazelcastInstance node) {
+            this.node = node;
+        }
+    }
 }