fix rackaware policy is ineffective after delete zk rack info

Author: fanjianye

pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/BookieRackAffinityMapping.java

@@ -121,8 +121,6 @@ public synchronized void setConf(Configuration conf) {
             store.registerListener(this::handleUpdates);
             racksWithHost = bookieMappingCache.get(BOOKIE_INFO_ROOT_PATH).get()
                     .orElseGet(BookiesRackConfiguration::new);
-            updateRacksWithHost(racksWithHost);
-            watchAvailableBookies();
             for (Map<String, BookieInfo> bookieMapping : racksWithHost.values()) {
                 for (String address : bookieMapping.keySet()) {
                     bookieAddressListLastTime.add(BookieId.parse(address));
@@ -132,6 +130,8 @@ public synchronized void setConf(Configuration conf) {
                             bookieAddressListLastTime);
                 }
             }
+            updateRacksWithHost(racksWithHost);
+            watchAvailableBookies();
         } catch (InterruptedException | ExecutionException | MetadataException e) {
             throw new RuntimeException(METADATA_STORE_INSTANCE + " failed to init BookieId list");
         }

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/RackAwareTest.java

@@ -66,6 +66,11 @@ public Object[][] forceMinRackNumProvider() {
         return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
     }
 
+    @DataProvider(name = "deleteRackFirstAfterBkClientInitialProvider")
+    public Object[][] deleteRackFirstAfterBkClientInitialProvider() {
+        return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
+    }
+
     @Override
     protected void configurePulsar(ServiceConfiguration config) throws Exception {
         // Start bookies with specific racks
@@ -201,7 +206,8 @@ public void testPlacementMinRackNumsPerWriteQuorum(boolean forceMinRackNums) thr
         }
     }
 
-    public void testRackUpdate() throws Exception {
+    @Test(dataProvider="deleteRackFirstAfterBkClientInitialProvider")
+    public void testRackUpdate(boolean isDeleteRackFirst) throws Exception {
         // 1. reset configurations for rack-aware
         cleanup();
         config = new ServiceConfiguration();
@@ -297,6 +303,14 @@ public void testRackUpdate() throws Exception {
             lh.close();
         }
 
+        // rackInfo is deleted first after bkClient initialize
+        if (isDeleteRackFirst) {
+            pulsar.getManagedLedgerClientFactory().close();
+            pulsar.getManagedLedgerClientFactory().initialize(pulsar.getConfig(),
+                    pulsar.getLocalMetadataStore(), pulsar.getBkClientFactory(), pulsar.getIoEventLoopGroup());
+            bkc = pulsar.getManagedLedgerClientFactory().getBookKeeperClient();
+        }
+
         // 6. remove rack-0
         for (int i = 0; i < NUM_BOOKIES / 2; i++) {
             String bookie = servers.get(i).getServer().getLocalAddress().toString();