You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/20 17:03:50 UTC
(pulsar) branch master updated: [fix][test] Fix flaky PulsarLedgerUnderreplicationManagerCloseTest.testCloseAggregatesFailuresAndRetriesOnlyUnreleasedLocks (#27013)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 5307d760114d [fix][test] Fix flaky PulsarLedgerUnderreplicationManagerCloseTest.testCloseAggregatesFailuresAndRetriesOnlyUnreleasedLocks (#27013)
5307d760114d is described below
commit 5307d760114dd23528dd33c16b5446f93de37abb
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Oct 20 20:03:43 2026 +0300
[fix][test] Fix flaky PulsarLedgerUnderreplicationManagerCloseTest.testCloseAggregatesFailuresAndRetriesOnlyUnreleasedLocks (#27013)
---
...lsarLedgerUnderreplicationManagerCloseTest.java | 23 ++++++++++++++--------
1 file changed, 15 insertions(+), 8 deletions(-)
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java
index e41f12942eac..063054e30fba 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java
@@ -30,6 +30,7 @@ import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -139,25 +140,31 @@ public class PulsarLedgerUnderreplicationManagerCloseTest extends BaseMetadataSt
try (var owner = newManager(observed, root);
var other = newManager(store, root)) {
acquireLedgers(owner);
+ // Deletes are issued from the store's callback threads in the order the ownership reads complete.
List<String> attemptedPaths = new ArrayList<>();
List<MetadataStoreException> failures = List.of(new MetadataStoreException("first failure"),
new MetadataStoreException("second failure"));
doAnswer(invocation -> {
String path = invocation.getArgument(0);
- attemptedPaths.add(path);
- if (attemptedPaths.size() <= failures.size()) {
- return CompletableFuture.failedFuture(failures.get(attemptedPaths.size() - 1));
+ int attempt;
+ synchronized (attemptedPaths) {
+ attempt = attemptedPaths.size();
+ attemptedPaths.add(path);
+ }
+ if (attempt < failures.size()) {
+ return CompletableFuture.failedFuture(failures.get(attempt));
}
return store.delete(path, invocation.getArgument(1));
}).when(observed).delete(anyString(), any());
+ // close() aggregates failures in held-lock order, which is independent of the delete order.
assertThatThrownBy(owner::close).isInstanceOf(UnavailableException.class)
.satisfies(error -> {
- assertThat(error.getCause()).isInstanceOf(ExecutionException.class)
- .hasCause(failures.get(0));
+ assertThat(error.getCause()).isInstanceOf(ExecutionException.class);
assertThat(error.getSuppressed()).hasSize(1);
- assertThat(error.getSuppressed()[0]).isInstanceOf(ExecutionException.class)
- .hasCause(failures.get(1));
+ assertThat(error.getSuppressed()[0]).isInstanceOf(ExecutionException.class);
+ assertThat(List.of(error.getCause().getCause(), error.getSuppressed()[0].getCause()))
+ .containsExactlyInAnyOrderElementsOf(failures);
});
assertThat(attemptedPaths).hasSize(LEDGERS.size());
for (long ledgerId : LEDGERS) {
@@ -166,7 +173,7 @@ public class PulsarLedgerUnderreplicationManagerCloseTest extends BaseMetadataSt
}
long reacquired = other.pollLedgerToRereplicate();
assertThat(reacquired).isIn(LEDGERS);
- List<String> retryPaths = new ArrayList<>();
+ List<String> retryPaths = new CopyOnWriteArrayList<>();
doAnswer(invocation -> {
String path = invocation.getArgument(0);
retryPaths.add(path);