[fix] [auto-recovery] Fix PulsarLedgerUnderreplicationManager notify …

…problem. (apache#21161)
ascentstream · Sep 18, 2023 · 8ff51eb · 8ff51eb
1 parent 1363777
commit 8ff51eb
Show file tree

Hide file tree

Showing 4 changed files with 371 additions and 26 deletions.
diff --git a/pulsar-metadata/pom.xml b/pulsar-metadata/pom.xml
@@ -57,7 +57,6 @@
       </exclusions>
     </dependency>
 
-    <!-- zookeeper server -->
     <dependency>
       <groupId>io.dropwizard.metrics</groupId>
       <artifactId>metrics-core</artifactId>
@@ -85,7 +84,6 @@
       </exclusions>
     </dependency>
 
-    <!-- zookeeper server -->
     <dependency>
       <groupId>org.xerial.snappy</groupId>
       <artifactId>snappy-java</artifactId>

diff --git a/.../main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java b/.../main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java
@@ -105,14 +105,17 @@ long getLedgerNodeVersion() {
     private final String urLockPath;
     private final String layoutPath;
     private final String lostBookieRecoveryDelayPath;
+    private final String replicationDisablePath;
     private final String checkAllLedgersCtimePath;
     private final String placementPolicyCheckCtimePath;
     private final String replicasCheckCtimePath;
 
     private final MetadataStoreExtended store;
 
-    private BookkeeperInternalCallbacks.GenericCallback<Void> replicationEnabledListener;
-    private BookkeeperInternalCallbacks.GenericCallback<Void> lostBookieRecoveryDelayListener;
+    private final List<BookkeeperInternalCallbacks.GenericCallback<Void>> replicationEnabledCallbacks =
+            new ArrayList<>();
+    private final List<BookkeeperInternalCallbacks.GenericCallback<Void>> lostBookieRecoveryDelayCallbacks =
+            new ArrayList<>();
 
     private static class PulsarUnderreplicatedLedger extends UnderreplicatedLedger {
         PulsarUnderreplicatedLedger(long ledgerId) {
@@ -139,6 +142,7 @@ public PulsarLedgerUnderreplicationManager(AbstractConfiguration<?> conf, Metada
         urLedgerPath = basePath + BookKeeperConstants.DEFAULT_ZK_LEDGERS_ROOT_PATH;
         urLockPath = basePath + '/' + BookKeeperConstants.UNDER_REPLICATION_LOCK;
         lostBookieRecoveryDelayPath = basePath + '/' + BookKeeperConstants.LOSTBOOKIERECOVERYDELAY_NODE;
+        replicationDisablePath = basePath + '/' + BookKeeperConstants.DISABLE_NODE;
         checkAllLedgersCtimePath = basePath + '/' + BookKeeperConstants.CHECK_ALL_LEDGERS_CTIME;
         placementPolicyCheckCtimePath = basePath + '/' + BookKeeperConstants.PLACEMENT_POLICY_CHECK_CTIME;
         replicasCheckCtimePath = basePath + '/' + BookKeeperConstants.REPLICAS_CHECK_CTIME;
@@ -232,17 +236,34 @@ private void handleNotification(Notification n) {
             synchronized (this) {
                 // Notify that there were some changes on the under-replicated z-nodes
                 notifyAll();
-
-                if (n.getType() == NotificationType.Deleted) {
-                    if (n.getPath().equals(basePath + '/' + BookKeeperConstants.DISABLE_NODE)) {
-                        log.info("LedgerReplication is enabled externally through MetadataStore, "
-                                + "since DISABLE_NODE ZNode is deleted");
-                        if (replicationEnabledListener != null) {
-                            replicationEnabledListener.operationComplete(0, null);
+                if (lostBookieRecoveryDelayPath.equals(n.getPath())) {
+                    final List<BookkeeperInternalCallbacks.GenericCallback<Void>> callbackList;
+                    synchronized (lostBookieRecoveryDelayCallbacks) {
+                        callbackList = new ArrayList<>(lostBookieRecoveryDelayCallbacks);
+                        lostBookieRecoveryDelayCallbacks.clear();
+                    }
+                    for (BookkeeperInternalCallbacks.GenericCallback<Void> callback : callbackList) {
+                        try {
+                            callback.operationComplete(0, null);
+                        } catch (Exception e) {
+                            log.warn("lostBookieRecoveryDelayCallbacks handle error", e);
                         }
-                    } else if (n.getPath().equals(lostBookieRecoveryDelayPath)) {
-                        if (lostBookieRecoveryDelayListener != null) {
-                            lostBookieRecoveryDelayListener.operationComplete(0, null);
+                    }
+                    return;
+                }
+                if (replicationDisablePath.equals(n.getPath()) && n.getType() == NotificationType.Deleted) {
+                    log.info("LedgerReplication is enabled externally through MetadataStore, "
+                            + "since DISABLE_NODE ZNode is deleted");
+                    final List<BookkeeperInternalCallbacks.GenericCallback<Void>> callbackList;
+                    synchronized (replicationEnabledCallbacks) {
+                        callbackList = new ArrayList<>(replicationEnabledCallbacks);
+                        replicationEnabledCallbacks.clear();
+                    }
+                    for (BookkeeperInternalCallbacks.GenericCallback<Void> callback : callbackList) {
+                        try {
+                            callback.operationComplete(0, null);
+                        } catch (Exception e) {
+                            log.warn("replicationEnabledCallbacks handle error", e);
                         }
                     }
                 }
@@ -688,8 +709,7 @@ public void disableLedgerReplication()
             log.debug("disableLedegerReplication()");
         }
         try {
-            String path = basePath + '/' + BookKeeperConstants.DISABLE_NODE;
-            store.put(path, "".getBytes(UTF_8), Optional.of(-1L))
+            store.put(replicationDisablePath, "".getBytes(UTF_8), Optional.of(-1L))
                     .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
             log.info("Auto ledger re-replication is disabled!");
         } catch (ExecutionException | TimeoutException ee) {
@@ -710,7 +730,7 @@ public void enableLedgerReplication()
             log.debug("enableLedegerReplication()");
         }
         try {
-            store.delete(basePath + '/' + BookKeeperConstants.DISABLE_NODE, Optional.empty())
+            store.delete(replicationDisablePath, Optional.empty())
                     .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
             log.info("Resuming automatic ledger re-replication");
         } catch (ExecutionException | TimeoutException ee) {
@@ -731,7 +751,7 @@ public boolean isLedgerReplicationEnabled()
             log.debug("isLedgerReplicationEnabled()");
         }
         try {
-            return !store.exists(basePath + '/' + BookKeeperConstants.DISABLE_NODE)
+            return !store.exists(replicationDisablePath)
                     .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
         } catch (ExecutionException | TimeoutException ee) {
             log.error("Error while checking the state of "
@@ -751,13 +771,11 @@ public void notifyLedgerReplicationEnabled(final BookkeeperInternalCallbacks.Gen
         if (log.isDebugEnabled()) {
             log.debug("notifyLedgerReplicationEnabled()");
         }
-
-        synchronized (this) {
-            replicationEnabledListener = cb;
+        synchronized (replicationEnabledCallbacks) {
+            replicationEnabledCallbacks.add(cb);
         }
-
         try {
-            if (!store.exists(basePath + '/' + BookKeeperConstants.DISABLE_NODE)
+            if (!store.exists(replicationDisablePath)
                     .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS)) {
                 log.info("LedgerReplication is enabled externally through metadata store, "
                         + "since DISABLE_NODE node is deleted");
@@ -851,8 +869,8 @@ public int getLostBookieRecoveryDelay() throws ReplicationException.UnavailableE
     public void notifyLostBookieRecoveryDelayChanged(BookkeeperInternalCallbacks.GenericCallback<Void> cb) throws
             ReplicationException.UnavailableException {
         log.debug("notifyLostBookieRecoveryDelayChanged()");
-        synchronized (this) {
-            lostBookieRecoveryDelayListener = cb;
+        synchronized (lostBookieRecoveryDelayCallbacks) {
+            lostBookieRecoveryDelayCallbacks.add(cb);
         }
         try {
             if (!store.exists(lostBookieRecoveryDelayPath).get(BLOCKING_CALL_TIMEOUT, MILLISECONDS)) {