BP-41 Add flag to enable/disable BookieAddressResolver

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java

@@ -497,8 +497,9 @@ public BookKeeper(ClientConfiguration conf, ZooKeeper zk, EventLoopGroup eventLo
             this.ownTimer = false;
         }
 
-        BookieAddressResolver bookieAddressResolver =
-                new DefaultBookieAddressResolver(metadataDriver.getRegistrationClient());
+        BookieAddressResolver bookieAddressResolver = conf.getBookieAddressResolverEnabled()
+                ? new DefaultBookieAddressResolver(metadataDriver.getRegistrationClient())
+                : new BookieAddressResolverDisabled();
         if (dnsResolver != null) {
             dnsResolver.setBookieAddressResolver(bookieAddressResolver);
         }

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieAddressResolverDisabled.java

@@ -0,0 +1,39 @@
+/**
+ * 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 lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.proto.BookieAddressResolver;
+
+/**
+ * Resolve legacy style BookieIDs to Network addresses.
+ */
+@Slf4j
+public final class BookieAddressResolverDisabled implements BookieAddressResolver {
+
+    public BookieAddressResolverDisabled() {
+    }
+
+    @Override
+    public BookieSocketAddress resolve(BookieId bookieId) {
+        return BookieSocketAddress.resolveLegacyBookieId(bookieId);
+    }
+
+}

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/DefaultBookieAddressResolver.java

@@ -57,7 +57,7 @@ public BookieSocketAddress resolve(BookieId bookieId) {
         } catch (BKException.BKBookieHandleNotAvailableException ex) {
             if (BookieSocketAddress.isDummyBookieIdForHostname(bookieId)) {
                 log.debug("Resolving dummy bookie Id {} using legacy bookie resolver", bookieId);
-                return BookieSocketAddress.resolveDummyBookieId(bookieId);
+                return BookieSocketAddress.resolveLegacyBookieId(bookieId);
             }
             log.info("Cannot resolve {}, bookie is unknown {}", bookieId, ex.toString());
             throw new BookieIdNotResolvedException(bookieId, ex);

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java

@@ -162,6 +162,7 @@ public class ClientConfiguration extends AbstractConfiguration<ClientConfigurati
     protected static final String READ_REORDER_THRESHOLD_PENDING_REQUESTS = "readReorderThresholdPendingRequests";
     protected static final String ENSEMBLE_PLACEMENT_POLICY_ORDER_SLOW_BOOKIES =
         "ensemblePlacementPolicyOrderSlowBookies";
+    protected static final String BOOKIE_ADDRESS_RESOLVER_ENABLED = "bookieAddressResolverEnabled";
 
     // Stats
     protected static final String ENABLE_TASK_EXECUTION_STATS = "enableTaskExecutionStats";
@@ -1288,6 +1289,33 @@ public ClientConfiguration setEnsemblePlacementPolicySlowBookies(boolean enabled
         return this;
     }
 
+    /**
+     * Whether to enable BookieAddressResolver.
+     *
+     * @return flag to enable/disable BookieAddressResolver.
+     */
+    public boolean getBookieAddressResolverEnabled() {
+        return getBoolean(BOOKIE_ADDRESS_RESOLVER_ENABLED, true);
+    }
+
+    /**
+     * Enable/Disable BookieAddressResolver.
+     *
+     * <p>
+     * If this flag is true, read bookie information from the metadata service (e.g. ZooKeeper) to resolve the address
+     * from each bookie ID. If all bookie IDs in the cluster are "address:port" or "hostname:port", you can set this
+     * flag to false to reduce requests to the metadata service.
+     * </p>
+     *
+     * @param enabled
+     *          flag to enable/disable BookieAddressResolver.
+     * @return client configuration.
+     */
+    public ClientConfiguration setBookieAddressResolverEnabled(boolean enabled) {
+        setProperty(BOOKIE_ADDRESS_RESOLVER_ENABLED, enabled);
+        return this;
+    }
+
     /**
      * Whether to enable recording task execution stats.
      *

bookkeeper-server/src/main/java/org/apache/bookkeeper/net/BookieSocketAddress.java

@@ -167,11 +167,10 @@ public static boolean isDummyBookieIdForHostname(BookieId bookieId) {
 
     /**
      * Use legacy resolver to resolve a bookieId.
-     * @param bookieId id supposed to be generated by
-     * {@link #createDummyBookieIdForHostname(java.lang.String)}
+     * @param bookieId legacy style bookie ID consisting of address (or hostname) and port
      * @return the BookieSocketAddress
      */
-    public static BookieSocketAddress resolveDummyBookieId(BookieId bookieId)
+    public static BookieSocketAddress resolveLegacyBookieId(BookieId bookieId)
             throws BookieAddressResolver.BookieIdNotResolvedException {
         return LEGACY_BOOKIEID_RESOLVER.resolve(bookieId);
     }

bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/ListBookiesCommand.java

@@ -26,6 +26,7 @@
 import java.util.Set;
 import lombok.Setter;
 import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.BookieAddressResolverDisabled;
 import org.apache.bookkeeper.client.DefaultBookieAddressResolver;
 import org.apache.bookkeeper.discover.RegistrationClient;
 import org.apache.bookkeeper.net.BookieId;
@@ -75,22 +76,27 @@ public static class Flags extends CliFlags {
     }
 
     @Override
-    protected void run(RegistrationClient regClient, Flags flags) throws Exception {
+    protected void run(RegistrationClient regClient, Flags flags, boolean bookieAddressResolverEnabled)
+            throws Exception {
         if (!flags.readwrite && !flags.readonly && !flags.all) {
             // case: no args is provided. list all the bookies by default.
             flags.readwrite = true;
             flags.readonly = true;
             flags.all = true;
         }
 
+        BookieAddressResolver bookieAddressResolver = bookieAddressResolverEnabled
+                ? new DefaultBookieAddressResolver(regClient)
+                : new BookieAddressResolverDisabled();
+
         boolean hasBookies = false;
         if (flags.readwrite) {
             Set<BookieId> bookies = result(
                 regClient.getWritableBookies()
             ).getValue();
             if (!bookies.isEmpty()) {
                 LOG.info("ReadWrite Bookies :");
-                printBookies(bookies, new DefaultBookieAddressResolver(regClient));
+                printBookies(bookies, bookieAddressResolver);
                 hasBookies = true;
             }
         }
@@ -100,7 +106,7 @@ protected void run(RegistrationClient regClient, Flags flags) throws Exception {
             ).getValue();
             if (!bookies.isEmpty()) {
                 LOG.info("Readonly Bookies :");
-                printBookies(bookies, new DefaultBookieAddressResolver(regClient));
+                printBookies(bookies, bookieAddressResolver);
                 hasBookies = true;
             }
         }
@@ -110,7 +116,7 @@ protected void run(RegistrationClient regClient, Flags flags) throws Exception {
             ).getValue();
             if (!bookies.isEmpty()) {
                 LOG.info("All Bookies :");
-                printBookies(bookies, new DefaultBookieAddressResolver(regClient));
+                printBookies(bookies, bookieAddressResolver);
                 hasBookies = true;
             }
         }

bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/helpers/DiscoveryCommand.java

@@ -56,7 +56,7 @@ protected boolean apply(ClientConfiguration clientConf, DiscoveryFlagsT cmdFlags
                     executor,
                     NullStatsLogger.INSTANCE,
                     Optional.empty());
-                run(driver.getRegistrationClient(), cmdFlags);
+                run(driver.getRegistrationClient(), cmdFlags, clientConf.getBookieAddressResolverEnabled());
                 return true;
             }
         } catch (Exception e) {
@@ -70,7 +70,7 @@ protected void run(BookKeeper bk, DiscoveryFlagsT cmdFlags) throws Exception {
         throw new IllegalStateException("It should never be called.");
     }
 
-    protected abstract void run(RegistrationClient regClient, DiscoveryFlagsT cmdFlags)
-        throws Exception;
+    protected abstract void run(RegistrationClient regClient, DiscoveryFlagsT cmdFlags,
+            boolean bookieAddressResolverEnabled) throws Exception;
 
 }

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieAddressResolverDisabledTest.java

@@ -0,0 +1,54 @@
+/*
+ *
+ * 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 org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.proto.BookieAddressResolver;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit test of {@link BookieAddressResolverDisabled}.
+ */
+public class BookieAddressResolverDisabledTest {
+
+    @Test
+    public void testResolve() {
+        BookieAddressResolver resolver = new BookieAddressResolverDisabled();
+
+        BookieSocketAddress addr1 = resolver.resolve(BookieId.parse("127.0.0.1:3181"));
+        Assert.assertEquals("127.0.0.1", addr1.getHostName());
+        Assert.assertEquals(3181, addr1.getPort());
+
+        BookieSocketAddress addr2 = resolver.resolve(BookieId.parse("localhost:3182"));
+        Assert.assertEquals("localhost", addr2.getHostName());
+        Assert.assertEquals(3182, addr2.getPort());
+
+        try {
+            resolver.resolve(BookieId.parse("foobar"));
+            Assert.fail("Non-legacy style bookie id should fail to resolve address");
+        } catch (Exception e) {
+            Assert.assertTrue(e instanceof BookieAddressResolver.BookieIdNotResolvedException);
+        }
+    }
+
+}

conf/bk_server.conf

@@ -960,6 +960,12 @@ zkEnableSecurity=false
 # acknowledged by bookkeeper.
 # enforceMinNumFaultDomainsForWrite=false
 
+# Whether to enable BookieAddressResolver.
+# If this flag is true, read bookie information from the metadata service (e.g. ZooKeeper) to resolve the address
+# from each bookie ID. If all bookie IDs in the cluster are "address:port" or "hostname:port", you can set this
+# flag to false to reduce requests to the metadata service.
+# bookieAddressResolverEnabled=true
+
 #############################################################################
 ## Auditor settings
 #############################################################################

stream/bk-grpc-name-resolver/src/main/java/org/apache/bookkeeper/grpc/resolver/BKRegistrationNameResolver.java

@@ -34,11 +34,13 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.stream.Collectors;
+import org.apache.bookkeeper.client.BookieAddressResolverDisabled;
 import org.apache.bookkeeper.client.DefaultBookieAddressResolver;
 import org.apache.bookkeeper.conf.ClientConfiguration;
 import org.apache.bookkeeper.meta.MetadataClientDriver;
 import org.apache.bookkeeper.meta.exceptions.MetadataException;
 import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.proto.BookieAddressResolver;
 import org.apache.bookkeeper.stats.NullStatsLogger;
 
 /**
@@ -53,7 +55,7 @@ class BKRegistrationNameResolver extends NameResolver {
     private Listener listener;
     private boolean shutdown;
     private boolean resolving;
-    private DefaultBookieAddressResolver bookieAddressResolver;
+    private BookieAddressResolver bookieAddressResolver;
 
     BKRegistrationNameResolver(MetadataClientDriver clientDriver,
                                URI serviceURI) {
@@ -81,7 +83,9 @@ public synchronized void start(Listener listener) {
         } catch (MetadataException e) {
             throw new RuntimeException("Failed to initialize registration client driver at " + serviceURI, e);
         }
-        this.bookieAddressResolver = new DefaultBookieAddressResolver(clientDriver.getRegistrationClient());
+        this.bookieAddressResolver = conf.getBookieAddressResolverEnabled()
+                ? new DefaultBookieAddressResolver(clientDriver.getRegistrationClient())
+                : new BookieAddressResolverDisabled();
 
         resolve();
     }

tools/ledger/src/test/java/org/apache/bookkeeper/tools/cli/helpers/ClientCommandTest.java

@@ -56,6 +56,7 @@ public void setup() throws Exception {
         this.serverConf.setMetadataServiceUri("zk://127.0.0.1/path/to/ledgers");
         mockConstruction(ClientConfiguration.class, (conf, context) -> {
             doReturn("zk://127.0.0.1/path/to/ledgers").when(conf).getMetadataServiceUri();
+            doReturn(true).when(conf).getBookieAddressResolverEnabled();
         });
         this.bkBuilder = mock(BookKeeperBuilder.class, CALLS_REAL_METHODS);
         mockStatic(BookKeeper.class).when(() ->

tools/ledger/src/test/java/org/apache/bookkeeper/tools/cli/helpers/ClientCommandTestBase.java

@@ -46,9 +46,10 @@ public abstract class ClientCommandTestBase extends CommandTestBase {
     public void setup() throws Exception {
         mockBk = mock(BookKeeper.class);
         mockConstruction(ClientConfiguration.class, withSettings().defaultAnswer(CALLS_REAL_METHODS),
-                        (mock, context) ->
-                        doReturn("zk://127.0.0.1/path/to/ledgers").when(mock).getMetadataServiceUri()
-                );
+                (mock, context) -> {
+                    doReturn("zk://127.0.0.1/path/to/ledgers").when(mock).getMetadataServiceUri();
+                    doReturn(true).when(mock).getBookieAddressResolverEnabled();
+                });
 
         mockStatic(BookKeeper.class);
         this.mockBkBuilder = mock(BookKeeperBuilder.class, CALLS_REAL_METHODS);

tools/ledger/src/test/java/org/apache/bookkeeper/tools/cli/helpers/CommandTestBase.java

@@ -125,6 +125,7 @@ protected void mockClientConfigurationConstruction() {
     protected void mockClientConfigurationConstruction(Consumer<ClientConfiguration> consumer) {
         mockConstruction(ClientConfiguration.class, (clientConfiguration, context) -> {
             doReturn("zk://127.0.0.1/path/to/ledgers").when(clientConfiguration).getMetadataServiceUri();
+            doReturn(true).when(clientConfiguration).getBookieAddressResolverEnabled();
             if (consumer != null) {
                 consumer.accept(clientConfiguration);
             }

tools/ledger/src/test/java/org/apache/bookkeeper/tools/cli/helpers/DiscoveryCommandTest.java

@@ -33,24 +33,26 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import org.apache.bookkeeper.conf.ClientConfiguration;
-import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.discover.RegistrationClient;
 import org.apache.bookkeeper.meta.MetadataClientDriver;
 import org.apache.bookkeeper.meta.MetadataDrivers;
 import org.apache.bookkeeper.stats.NullStatsLogger;
 import org.apache.bookkeeper.tools.framework.CliFlags;
 import org.junit.Before;
-import org.junit.Test;
+import org.junit.experimental.theories.DataPoint;
+import org.junit.experimental.theories.Theories;
+import org.junit.experimental.theories.Theory;
+import org.junit.runner.RunWith;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
 /**
  * Unit test of {@link DiscoveryCommand}.
  */
+@RunWith(Theories.class)
 public class DiscoveryCommandTest {
 
     private DiscoveryCommand<CliFlags> cmd;
-    private ServerConfiguration serverConf;
     private ClientConfiguration clientConf;
     private RegistrationClient regClient;
     private MetadataClientDriver clientDriver;
@@ -62,17 +64,23 @@ public void setup() throws Exception {
 
         this.cmd = mock(DiscoveryCommand.class, CALLS_REAL_METHODS);
 
-        this.serverConf = new ServerConfiguration();
-        this.serverConf.setMetadataServiceUri("zk://127.0.0.1/path/to/ledgers");
+        this.clientConf = new ClientConfiguration();
+        this.clientConf.setMetadataServiceUri("zk://127.0.0.1/path/to/ledgers");
         this.executor = mock(ScheduledExecutorService.class);
         this.regClient = mock(RegistrationClient.class);
         this.clientDriver = mock(MetadataClientDriver.class);
         when(clientDriver.getRegistrationClient())
             .thenReturn(regClient);
     }
 
-    @Test
-    public void testRun() throws Exception {
+    @DataPoint
+    public static final boolean BOOKIE_ADDR_RESOLVER_ENABLED = true;
+    @DataPoint
+    public static final boolean BOOKIE_ADDR_RESOLVER_DISABLED = false;
+    @Theory
+    public void testRun(boolean bookieAddressResolverEnabled) throws Exception {
+        clientConf.setBookieAddressResolverEnabled(bookieAddressResolverEnabled);
+
         try (final MockedStatic<Executors> executorsMockedStatic = Mockito.mockStatic(Executors.class);
              final MockedStatic<MetadataDrivers> mdriversMockedStatic = Mockito.mockStatic(MetadataDrivers.class);) {
             executorsMockedStatic
@@ -81,8 +89,8 @@ public void testRun() throws Exception {
                     .thenReturn(clientDriver);
 
             CliFlags cliFlags = new CliFlags();
-            assertTrue(cmd.apply(serverConf, cliFlags));
-            verify(cmd, times(1)).run(eq(regClient), same(cliFlags));
+            assertTrue(cmd.apply(clientConf, cliFlags));
+            verify(cmd, times(1)).run(eq(regClient), same(cliFlags), eq(bookieAddressResolverEnabled));
             verify(clientDriver, times(1))
                 .initialize(
                         any(ClientConfiguration.class), eq(executor),