fix race condition (apache#13007)

lujp · Nov 29, 2021 · 9add033 · 9add033
1 parent bb2c934
commit 9add033
Showing 1 changed file with 10 additions and 7 deletions.
diff --git a/...r-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java b/...r-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java
@@ -192,13 +192,16 @@ private Optional<MetadataStoreException> programmedFailure(OperationType op, Str
         if (ex != null) {
             return Optional.of(ex);
         }
-        Optional<Failure> failure = failures.stream()
-                .filter(f -> f.predicate.test(op, path))
-                .findFirst();
-        if (failure.isPresent()) {
-            failures.remove(failure.get());
+        while (true) {
+            Optional<Failure> failure = failures.stream().filter(f -> f.predicate.test(op, path)).findFirst();
+            if (failure.isPresent()) {
+                if (failures.remove(failure.get())) {
+                    return failure.map(Failure::getException);
+                }
+                // failure is taken by other threads. Retry.
+            } else {
+                return Optional.empty();
+            }
         }
-
-        return failure.map(Failure::getException);
     }
 }