[fix][standalone] correctly delete bookie registration znode (apache#…

…23497) Signed-off-by: Zixuan Liu <[email protected]> (cherry picked from commit ebb3cb5)
ascentstream · Nov 6, 2024 · d2d9b94 · d2d9b94
1 parent 5a63ebd
commit d2d9b94
Showing 1 changed file with 15 additions and 11 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java b/pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java
@@ -297,22 +297,16 @@ private void runBookies(ServerConfiguration baseConf) throws Exception {
             }
 
             int bookiePort = portManager.get();
-
+            String bookieId = "bk" + i + "test";
             // Ensure registration Z-nodes are cleared when standalone service is restarted ungracefully
-            String registrationZnode = String.format("/ledgers/available/%s:%d",
-                    baseConf.getAdvertisedAddress(), bookiePort);
-            if (zkc.exists(registrationZnode, null) != null) {
-                try {
-                    zkc.delete(registrationZnode, -1);
-                } catch (NoNodeException nne) {
-                    // Ignore if z-node was just expired
-                }
-            }
+            deleteBookieRegistrationZnode(
+                    String.format("/ledgers/available/%s:%d", baseConf.getAdvertisedAddress(), bookiePort));
+            deleteBookieRegistrationZnode(String.format("/ledgers/available/%s", bookieId));
 
             bsConfs[i] = new ServerConfiguration(baseConf);
             // override settings
             bsConfs[i].setBookiePort(bookiePort);
-            bsConfs[i].setBookieId("bk" + i + "test");
+            bsConfs[i].setBookieId(bookieId);
             String zkServers = "127.0.0.1:" + zkPort;
             String metadataServiceUriStr = "zk://" + zkServers + "/ledgers";
 
@@ -327,6 +321,16 @@ private void runBookies(ServerConfiguration baseConf) throws Exception {
         }
     }
 
+    private void deleteBookieRegistrationZnode(String registrationZnode) throws InterruptedException, KeeperException {
+        if (zkc.exists(registrationZnode, null) != null) {
+            try {
+                zkc.delete(registrationZnode, -1);
+            } catch (NoNodeException nne) {
+                // Ignore if z-node was just expired
+            }
+        }
+    }
+
     public void runStreamStorage(CompositeConfiguration conf) throws Exception {
         String zkServers = "127.0.0.1:" + zkPort;
         String metadataServiceUriStr = "zk://" + zkServers + "/ledgers";