diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java index 663ad63e50b..76ef7bd2654 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java @@ -168,6 +168,9 @@ public class ClientConfiguration extends AbstractConfiguration> bookieServiceInfoCache = new ConcurrentHashMap<>(); - + private final Watcher bookieServiceInfoCacheInvalidation; + private final boolean bookieAddressTracking; // registration paths private final String bookieRegistrationPath; private final String bookieAllRegistrationPath; @@ -190,10 +191,17 @@ public void close() { public ZKRegistrationClient(ZooKeeper zk, String ledgersRootPath, - ScheduledExecutorService scheduler) { + ScheduledExecutorService scheduler, + boolean bookieAddressTracking) { this.zk = zk; this.scheduler = scheduler; - + // Following Bookie Network Address Changes is an expensive operation + // as it requires additional ZooKeeper watches + // we can disable this feature, in case the BK cluster has only + // static addresses + this.bookieAddressTracking = bookieAddressTracking; + this.bookieServiceInfoCacheInvalidation = bookieAddressTracking + ? new BookieServiceInfoCacheInvalidationWatcher() : null; this.bookieRegistrationPath = ledgersRootPath + "/" + AVAILABLE_NODE; this.bookieAllRegistrationPath = ledgersRootPath + "/" + COOKIE_NODE; this.bookieReadonlyRegistrationPath = this.bookieRegistrationPath + "/" + READONLY; @@ -204,6 +212,10 @@ public void close() { // no-op } + public boolean isBookieAddressTracking() { + return bookieAddressTracking; + } + public ZooKeeper getZk() { return zk; } @@ -243,12 +255,13 @@ public CompletableFuture> getBookieServiceInfo(Book * @param bookieId * @return an handle to the result of the operation. */ - private CompletableFuture> readBookieServiceInfo(BookieId bookieId) { + private CompletableFuture> readBookieServiceInfoAsync(BookieId bookieId) { String pathAsWritable = bookieRegistrationPath + "/" + bookieId; String pathAsReadonly = bookieReadonlyRegistrationPath + "/" + bookieId; CompletableFuture> promise = new CompletableFuture<>(); - zk.getData(pathAsWritable, null, (int rc, String path, Object o, byte[] bytes, Stat stat) -> { + zk.getData(pathAsWritable, bookieServiceInfoCacheInvalidation, + (int rc, String path, Object o, byte[] bytes, Stat stat) -> { if (KeeperException.Code.OK.intValue() == rc) { try { BookieServiceInfo bookieServiceInfo = deserializeBookieServiceInfo(bookieId, bytes); @@ -265,7 +278,8 @@ private CompletableFuture> readBookieServiceInfo(Bo } } else if (KeeperException.Code.NONODE.intValue() == rc) { // not found, looking for a readonly bookie - zk.getData(pathAsReadonly, null, (int rc2, String path2, Object o2, byte[] bytes2, Stat stat2) -> { + zk.getData(pathAsReadonly, bookieServiceInfoCacheInvalidation, + (int rc2, String path2, Object o2, byte[] bytes2, Stat stat2) -> { if (KeeperException.Code.OK.intValue() == rc2) { try { BookieServiceInfo bookieServiceInfo = deserializeBookieServiceInfo(bookieId, bytes2); @@ -344,7 +358,7 @@ private CompletableFuture>> getChildren(String regPath, for (BookieId id : bookies) { // update the cache for new bookies if (!bookieServiceInfoCache.containsKey(id)) { - bookieInfoUpdated.add(readBookieServiceInfo(id)); + bookieInfoUpdated.add(readBookieServiceInfoAsync(id)); } } if (bookieInfoUpdated.isEmpty()) { @@ -447,4 +461,46 @@ private static HashSet convertToBookieAddresses(List children) return newBookieAddrs; } + private static BookieId stripBookieIdFromPath(String path) { + final int slash = path.lastIndexOf('/'); + if (slash >= 0) { + try { + return BookieId.parse(path.substring(slash + 1)); + } catch (IllegalArgumentException e) { + log.warn("Cannot decode bookieId from {}", path, e); + } + } + return null; + } + + private class BookieServiceInfoCacheInvalidationWatcher implements Watcher { + + @Override + public void process(WatchedEvent we) { + log.debug("zk event {} for {} state {}", we.getType(), we.getPath(), we.getState()); + if (we.getState() == KeeperState.Expired) { + log.info("zk session expired, invalidating cache"); + bookieServiceInfoCache.clear(); + return; + } + BookieId bookieId = stripBookieIdFromPath(we.getPath()); + if (bookieId == null) { + return; + } + switch (we.getType()) { + case NodeDeleted: + log.info("Invalidate cache for {}", bookieId); + bookieServiceInfoCache.remove(bookieId); + break; + case NodeDataChanged: + log.info("refresh cache for {}", bookieId); + readBookieServiceInfoAsync(bookieId); + break; + default: + log.debug("ignore cache event {} for {}", we.getType(), bookieId); + break; + } + } + } + } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java index e8eea4939dc..df5499b80a8 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java @@ -515,7 +515,8 @@ public boolean nukeExistingCluster() throws Exception { try (RegistrationClient regClient = new ZKRegistrationClient( zk, ledgersRootPath, - null + null, + false )) { if (availableNodeExists) { Collection rwBookies = FutureUtils diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriver.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriver.java index 99b942771da..638f39066f8 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriver.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriver.java @@ -51,6 +51,7 @@ public class ZKMetadataClientDriver ClientConfiguration clientConf; ScheduledExecutorService scheduler; RegistrationClient regClient; + boolean bookieAddressTracking = true; @Override public synchronized MetadataClientDriver initialize(ClientConfiguration conf, @@ -70,6 +71,7 @@ public synchronized MetadataClientDriver initialize(ClientConfiguration conf, this.statsLogger = statsLogger; this.clientConf = conf; this.scheduler = scheduler; + this.bookieAddressTracking = conf.getEnableBookieAddressTracking(); return this; } @@ -79,7 +81,8 @@ public synchronized RegistrationClient getRegistrationClient() { regClient = new ZKRegistrationClient( zk, ledgersRootPath, - scheduler); + scheduler, + bookieAddressTracking); } return regClient; } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieNetworkAddressChangeTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieNetworkAddressChangeTest.java new file mode 100644 index 00000000000..977eb8cc308 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieNetworkAddressChangeTest.java @@ -0,0 +1,178 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.bookkeeper.client; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.client.BKException.BKBookieHandleNotAvailableException; +import org.apache.bookkeeper.client.api.BookKeeper; +import org.apache.bookkeeper.client.api.LedgerEntries; +import org.apache.bookkeeper.client.api.ReadHandle; +import org.apache.bookkeeper.client.api.WriteHandle; +import org.apache.bookkeeper.conf.ClientConfiguration; +import org.apache.bookkeeper.conf.ServerConfiguration; +import org.apache.bookkeeper.discover.ZKRegistrationClient; +import org.apache.bookkeeper.test.BookKeeperClusterTestCase; +import org.apache.bookkeeper.util.PortManager; +import org.junit.Test; + +/** + * Tests of the main BookKeeper client and the BP-41 bookieAddressTracking feature. + */ +@Slf4j +public class BookieNetworkAddressChangeTest extends BookKeeperClusterTestCase { + + public BookieNetworkAddressChangeTest() { + super(1); + this.useUUIDasBookieId = true; + } + + @Test + public void testFollowBookieAddressChange() throws Exception { + ClientConfiguration conf = new ClientConfiguration(); + conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri()); + try (BookKeeper bkc = BookKeeper.newBuilder(conf) + .build();) { + long lId; + try (WriteHandle h = bkc + .newCreateLedgerOp() + .withAckQuorumSize(1) + .withEnsembleSize(1) + .withWriteQuorumSize(1) + .withPassword(new byte[0]) + .execute() + .get();) { + lId = h.getId(); + h.append("foo".getBytes("utf-8")); + } + + // restart bookie, change port + // on metadata we have a bookieId, not the network address + ServerConfiguration thisServerConf = new ServerConfiguration(baseConf); + thisServerConf.setBookiePort(PortManager.nextFreePort()); + restartBookies(thisServerConf); + + try (ReadHandle h = bkc + .newOpenLedgerOp() + .withLedgerId(lId) + .withRecovery(true) + .withPassword(new byte[0]) + .execute() + .get()) { + assertEquals(0, h.getLastAddConfirmed()); + try (LedgerEntries entries = h.read(0, 0);) { + assertEquals("foo", new String(entries.getEntry(0).getEntryBytes(), "utf-8")); + } + } + } + } + + @Test + public void testFollowBookieAddressChangeTrckingDisabled() throws Exception { + ClientConfiguration conf = new ClientConfiguration(); + conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri()); + conf.setEnableBookieAddressTracking(false); + try (BookKeeper bkc = BookKeeper.newBuilder(conf) + .build();) { + long lId; + try (WriteHandle h = bkc + .newCreateLedgerOp() + .withAckQuorumSize(1) + .withEnsembleSize(1) + .withWriteQuorumSize(1) + .withPassword(new byte[0]) + .execute() + .get();) { + lId = h.getId(); + h.append("foo".getBytes("utf-8")); + } + + // restart bookie, change port + // on metadata we have a bookieId, not the network address + ServerConfiguration thisServerConf = new ServerConfiguration(baseConf); + thisServerConf.setBookiePort(PortManager.nextFreePort()); + restartBookies(thisServerConf); + + try (ReadHandle h = bkc + .newOpenLedgerOp() + .withLedgerId(lId) + .withRecovery(true) + .withPassword(new byte[0]) + .execute() + .get()) { + try (LedgerEntries entries = h.read(0, 0);) { + fail("Should not be able to connect to the bookie with Bookie Address Tracking Disabled"); + } catch (BKBookieHandleNotAvailableException expected) { + } + } + } + } + + @Test + public void testFollowBookieAddressChangeZkSessionExpire() throws Exception { + ClientConfiguration conf = new ClientConfiguration(); + conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri()); + try (BookKeeper bkc = BookKeeper.newBuilder(conf) + .build();) { + long lId; + try (WriteHandle h = bkc + .newCreateLedgerOp() + .withAckQuorumSize(1) + .withEnsembleSize(1) + .withWriteQuorumSize(1) + .withPassword(new byte[0]) + .execute() + .get();) { + lId = h.getId(); + h.append("foo".getBytes("utf-8")); + } + + log.error("expiring ZK session!"); + // expire zk session + ZKRegistrationClient regClient = (ZKRegistrationClient) ((org.apache.bookkeeper.client.BookKeeper) bkc) + .getMetadataClientDriver() + .getRegistrationClient(); + + regClient.getZk().getTestable().injectSessionExpiration(); + + // restart bookie, change port + // on metadata we have a bookieId, not the network address + ServerConfiguration thisServerConf = new ServerConfiguration(baseConf); + thisServerConf.setBookiePort(PortManager.nextFreePort()); + restartBookies(thisServerConf); + + try (ReadHandle h = bkc + .newOpenLedgerOp() + .withLedgerId(lId) + .withRecovery(true) + .withPassword(new byte[0]) + .execute() + .get()) { + assertEquals(0, h.getLastAddConfirmed()); + try (LedgerEntries entries = h.read(0, 0);) { + assertEquals("foo", new String(entries.getEntry(0).getEntryBytes(), "utf-8")); + } + } + } + } +} \ No newline at end of file diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClient.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/AbstractTestZkRegistrationClient.java similarity index 96% rename from bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClient.java rename to bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/AbstractTestZkRegistrationClient.java index a8d4f1dd479..e09a6350b16 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClient.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/AbstractTestZkRegistrationClient.java @@ -83,7 +83,9 @@ @RunWith(PowerMockRunner.class) @PrepareForTest({ ZKRegistrationClient.class, ZkUtils.class }) @Slf4j -public class TestZkRegistrationClient extends MockZooKeeperTestCase { +public abstract class AbstractTestZkRegistrationClient extends MockZooKeeperTestCase { + + @Rule public final TestName runtime = new TestName(); @@ -96,6 +98,13 @@ public class TestZkRegistrationClient extends MockZooKeeperTestCase { private ScheduledExecutorService mockExecutor; private MockExecutorController controller; + private final boolean bookieAddressChangeTracking; + + public AbstractTestZkRegistrationClient(boolean bookieAddressChangeTracking) { + this.bookieAddressChangeTracking = bookieAddressChangeTracking; + } + + @Override @Before public void setup() throws Exception { @@ -114,8 +123,10 @@ public void setup() throws Exception { this.zkRegistrationClient = new ZKRegistrationClient( mockZk, ledgersPath, - mockExecutor + mockExecutor, + bookieAddressChangeTracking ); + assertEquals(bookieAddressChangeTracking, zkRegistrationClient.isBookieAddressTracking()); } @After @@ -136,23 +147,23 @@ private static Set prepareNBookies(int num) { private void prepareReadBookieServiceInfo(BookieId address, boolean readonly) throws Exception { if (readonly) { mockZkGetData(regPath + "/" + address.toString(), - false, + zkRegistrationClient.isBookieAddressTracking(), Code.NONODE.intValue(), new byte[] {}, new Stat()); mockZkGetData(regReadonlyPath + "/" + address.toString(), - false, + zkRegistrationClient.isBookieAddressTracking(), Code.OK.intValue(), new byte[] {}, new Stat()); } else { mockZkGetData(regPath + "/" + address.toString(), - false, + zkRegistrationClient.isBookieAddressTracking(), Code.OK.intValue(), new byte[] {}, new Stat()); mockZkGetData(regReadonlyPath + "/" + address.toString(), - false, + zkRegistrationClient.isBookieAddressTracking(), Code.NONODE.intValue(), new byte[] {}, new Stat()); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClientWithBookieAddressTracking.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClientWithBookieAddressTracking.java new file mode 100644 index 00000000000..f9b6de342c2 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClientWithBookieAddressTracking.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.bookkeeper.discover; + +/** + * Unit test of {@link RegistrationClient} with Bookie Address Tracking feature. + */ +public class TestZkRegistrationClientWithBookieAddressTracking extends AbstractTestZkRegistrationClient { + + public TestZkRegistrationClientWithBookieAddressTracking() { + super(true); + } + +} diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClientWithoutBookieAddressTracking.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClientWithoutBookieAddressTracking.java new file mode 100644 index 00000000000..39ed8eacc81 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClientWithoutBookieAddressTracking.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + +package org.apache.bookkeeper.discover; + +/** + * Unit test of {@link RegistrationClient} without Bookie Address Tracking feature. + */ +public class TestZkRegistrationClientWithoutBookieAddressTracking extends AbstractTestZkRegistrationClient { + + public TestZkRegistrationClientWithoutBookieAddressTracking() { + super(false); + } + +} diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriverTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriverTest.java index 626a055815c..75aadef7955 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriverTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriverTest.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; @@ -72,8 +73,8 @@ public void testGetRegClient() throws Exception { ZKRegistrationClient mockRegClient = PowerMockito.mock(ZKRegistrationClient.class); PowerMockito.whenNew(ZKRegistrationClient.class) - .withParameterTypes(ZooKeeper.class, String.class, ScheduledExecutorService.class) - .withArguments(any(ZooKeeper.class), anyString(), any(ScheduledExecutorService.class)) + .withParameterTypes(ZooKeeper.class, String.class, ScheduledExecutorService.class, Boolean.TYPE) + .withArguments(any(ZooKeeper.class), anyString(), any(ScheduledExecutorService.class), anyBoolean()) .thenReturn(mockRegClient); RegistrationClient client = driver.getRegistrationClient(); @@ -81,7 +82,7 @@ public void testGetRegClient() throws Exception { assertSame(mockRegClient, driver.regClient); PowerMockito.verifyNew(ZKRegistrationClient.class, times(1)) - .withArguments(eq(mockZkc), eq(ledgersRootPath), eq(mockExecutor)); + .withArguments(eq(mockZkc), eq(ledgersRootPath), eq(mockExecutor), anyBoolean()); driver.close(); verify(mockRegClient, times(1)).close(); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java index 21568591e7a..a52a30489bf 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java @@ -614,12 +614,17 @@ public void restartBookies(ServerConfiguration newConf) Thread.sleep(1000); // restart them to ensure we can't for (ServerConfiguration conf : bsConfs) { - // ensure the bookie port is loaded correctly + String bookieId = conf.getBookieId(); + // ensure the bookie id or port is loaded correctly int port = conf.getBookiePort(); if (null != newConf) { conf.loadConf(newConf); } - conf.setBookiePort(port); + if (bookieId != null) { + conf.setBookieId(bookieId); + } else { + conf.setBookiePort(port); + } bs.add(startBookie(conf)); } } diff --git a/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/RegistrationServiceProvider.java b/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/RegistrationServiceProvider.java index f1cda91e710..03bf23d1a53 100644 --- a/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/RegistrationServiceProvider.java +++ b/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/RegistrationServiceProvider.java @@ -30,6 +30,7 @@ import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.component.AbstractLifecycleComponent; +import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.bookkeeper.conf.ServerConfiguration; import org.apache.bookkeeper.discover.RegistrationClient; import org.apache.bookkeeper.discover.ZKRegistrationClient; @@ -52,6 +53,7 @@ public class RegistrationServiceProvider private final RetryPolicy bkZkRetryPolicy; private final String regPath; private final ScheduledExecutorService regExecutor; + private final boolean bookieAddresschangeTracking; private ZooKeeperClient zkClient; private RegistrationClient client; @@ -67,6 +69,8 @@ public RegistrationServiceProvider(ServerConfiguration bkServerConf, Integer.MAX_VALUE); this.regExecutor = Executors.newSingleThreadScheduledExecutor( new ThreadFactoryBuilder().setNameFormat("registration-service-provider-scheduler").build()); + ClientConfiguration clientConfiguration = new ClientConfiguration(bkServerConf); + this.bookieAddresschangeTracking = clientConfiguration.getEnableBookieAddressTracking(); } @Override @@ -101,7 +105,7 @@ protected void doStart() { log.error("Failed to create zookeeper client to {}", zkServers, e); throw e; } - client = new ZKRegistrationClient(zkClient, regPath, regExecutor); + client = new ZKRegistrationClient(zkClient, regPath, regExecutor, bookieAddresschangeTracking); } }