Fix: LockManagerTest.updateValue is flaky (apache#13911)

drianpauL · Jan 25, 2022 · 78827be · 78827be
1 parent 15e7369
commit 78827be
Showing 1 changed file with 16 additions and 10 deletions.
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java
@@ -168,21 +168,24 @@ public CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Long>
 
             long now = System.currentTimeMillis();
 
+            CompletableFuture<Stat> future = new CompletableFuture<>();
             if (hasVersion && expectedVersion == -1) {
                 Value newValue = new Value(0, data, now, now, options.contains(CreateOption.Ephemeral));
                 Value existingValue = map.putIfAbsent(path, newValue);
                 if (existingValue != null) {
-                    return FutureUtils.exception(new BadVersionException(""));
+                    execute(() -> future.completeExceptionally(new BadVersionException("")), future);
                 } else {
                     receivedNotification(new Notification(NotificationType.Created, path));
                     notifyParentChildrenChanged(path);
-                    return FutureUtils.value(new Stat(path, 0, now, now, newValue.isEphemeral(), true));
+                    String finalPath = path;
+                    execute(() -> future.complete(new Stat(finalPath, 0, now, now, newValue.isEphemeral(),
+                            true)), future);
                 }
             } else {
                 Value existingValue = map.get(path);
                 long existingVersion = existingValue != null ? existingValue.version : -1;
                 if (hasVersion && expectedVersion != existingVersion) {
-                    return FutureUtils.exception(new BadVersionException(""));
+                    execute(() -> future.completeExceptionally(new BadVersionException("")), future);
                 } else {
                     long newVersion = existingValue != null ? existingValue.version + 1 : 0;
                     long createdTimestamp = existingValue != null ? existingValue.createdTimestamp : now;
@@ -196,12 +199,13 @@ public CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Long>
                     if (type == NotificationType.Created) {
                         notifyParentChildrenChanged(path);
                     }
-                    return FutureUtils
-                            .value(new Stat(path, newValue.version, newValue.createdTimestamp,
-                                    newValue.modifiedTimestamp,
-                                    false, true));
+                    String finalPath = path;
+                    execute(() -> future.complete(new Stat(finalPath, newValue.version, newValue.createdTimestamp,
+                            newValue.modifiedTimestamp,
+                            false, true)), future);
                 }
             }
+            return future;
         }
     }
 
@@ -211,18 +215,20 @@ public CompletableFuture<Void> storeDelete(String path, Optional<Long> optExpect
             return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
         }
         synchronized (map) {
+            CompletableFuture<Void> future = new CompletableFuture<>();
             Value value = map.get(path);
             if (value == null) {
-                return FutureUtils.exception(new NotFoundException(""));
+                execute(() -> future.completeExceptionally(new NotFoundException("")), future);
             } else if (optExpectedVersion.isPresent() && optExpectedVersion.get() != value.version) {
-                return FutureUtils.exception(new BadVersionException(""));
+                execute(() -> future.completeExceptionally(new BadVersionException("")), future);
             } else {
                 map.remove(path);
                 receivedNotification(new Notification(NotificationType.Deleted, path));
 
                 notifyParentChildrenChanged(path);
-                return FutureUtils.value(null);
+                execute(() -> future.complete(null), future);
             }
+            return future;
         }
     }
 }