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/06/22 14:13:00 UTC
(pulsar) branch master updated: [fix][meta] Run ledger-underreplication notification callbacks off the metadata-store listener thread (#26065)
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 cc96d1c385b0 [fix][meta] Run ledger-underreplication notification callbacks off the metadata-store listener thread (#26065)
cc96d1c385b0 is described below
commit cc96d1c385b0bc8e930579adb9aea742feedd191
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Jun 22 07:12:54 2026 -0700
[fix][meta] Run ledger-underreplication notification callbacks off the metadata-store listener thread (#26065)
---
.../PulsarLedgerUnderreplicationManager.java | 37 +++++++++++++++-------
1 file changed, 25 insertions(+), 12 deletions(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java
index 38d7a5cd54a6..1c9648af175b 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java
@@ -22,6 +22,7 @@ import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.pulsar.metadata.bookkeeper.AbstractMetadataDriver.BLOCKING_CALL_TIMEOUT;
import com.google.common.base.Joiner;
+import io.netty.util.concurrent.DefaultThreadFactory;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -37,6 +38,8 @@ import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.function.Predicate;
import java.util.regex.Matcher;
@@ -116,6 +119,11 @@ public class PulsarLedgerUnderreplicationManager implements LedgerUnderreplicati
private final List<BookkeeperInternalCallbacks.GenericCallback<Void>> lostBookieRecoveryDelayCallbacks =
new ArrayList<>();
+ // Registered callbacks can perform synchronous metadata-store reads, so run them on a dedicated
+ // single-threaded executor instead of the metadata-store notification thread (and outside the lock).
+ private final ExecutorService notificationCallbackExecutor =
+ Executors.newSingleThreadExecutor(new DefaultThreadFactory("pulsar-underreplication-notification"));
+
private static class PulsarUnderreplicatedLedger extends UnderreplicatedLedger {
PulsarUnderreplicatedLedger(long ledgerId) {
super(ledgerId);
@@ -246,13 +254,15 @@ public class PulsarLedgerUnderreplicationManager implements LedgerUnderreplicati
callbackList = new ArrayList<>(lostBookieRecoveryDelayCallbacks);
lostBookieRecoveryDelayCallbacks.clear();
}
- for (BookkeeperInternalCallbacks.GenericCallback<Void> callback : callbackList) {
- try {
- callback.operationComplete(0, null);
- } catch (Exception e) {
- log.warn().exception(e).log("lostBookieRecoveryDelayCallbacks handle error");
+ notificationCallbackExecutor.execute(() -> {
+ for (BookkeeperInternalCallbacks.GenericCallback<Void> callback : callbackList) {
+ try {
+ callback.operationComplete(0, null);
+ } catch (Exception e) {
+ log.warn().exception(e).log("lostBookieRecoveryDelayCallbacks handle error");
+ }
}
- }
+ });
return;
}
if (replicationDisablePath.equals(n.getPath()) && n.getType() == NotificationType.Deleted) {
@@ -263,13 +273,15 @@ public class PulsarLedgerUnderreplicationManager implements LedgerUnderreplicati
callbackList = new ArrayList<>(replicationEnabledCallbacks);
replicationEnabledCallbacks.clear();
}
- for (BookkeeperInternalCallbacks.GenericCallback<Void> callback : callbackList) {
- try {
- callback.operationComplete(0, null);
- } catch (Exception e) {
- log.warn().exception(e).log("replicationEnabledCallbacks handle error");
+ notificationCallbackExecutor.execute(() -> {
+ for (BookkeeperInternalCallbacks.GenericCallback<Void> callback : callbackList) {
+ try {
+ callback.operationComplete(0, null);
+ } catch (Exception e) {
+ log.warn().exception(e).log("replicationEnabledCallbacks handle error");
+ }
}
- }
+ });
}
}
}
@@ -670,6 +682,7 @@ public class PulsarLedgerUnderreplicationManager implements LedgerUnderreplicati
@Override
public void close() throws ReplicationException.UnavailableException {
log.debug("close()");
+ notificationCallbackExecutor.shutdownNow();
try {
for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
store.delete(e.getValue().getLockPath(), Optional.empty())