jiazhai · Sep 19, 2020 · Sep 8, 2020
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
@@ -85,6 +85,7 @@
 import org.apache.bookkeeper.net.BookieSocketAddress;
 import org.apache.bookkeeper.net.DNS;
 import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
+import org.apache.bookkeeper.proto.SimpleBookieServiceInfoProvider;
 import org.apache.bookkeeper.stats.NullStatsLogger;
 import org.apache.bookkeeper.stats.StatsLogger;
 import org.apache.bookkeeper.util.BookKeeperConstants;
@@ -329,6 +330,9 @@ static List<BookieId> possibleBookieIds(ServerConfiguration conf)
             if (null != conf.getAdvertisedAddress()) {
                 addresses.add(getBookieAddress(conf).toBookieId());
             }
+            if (null != conf.getBookieId()) {
+                addresses.add(BookieId.parse(conf.getBookieId()));
+            }
         } catch (UnknownHostException e) {
             throw new UnknownBookieIdException(e);
         }
@@ -534,7 +538,11 @@ private static void verifyDirsForStorageExpansion(
     }
 
     public static BookieId getBookieId(ServerConfiguration conf) throws UnknownHostException {
-         return getBookieAddress(conf).toBookieId();
+        String customBookieId = conf.getBookieId();
+        if (customBookieId != null) {
+            return BookieId.parse(customBookieId);
+        }
+        return getBookieAddress(conf).toBookieId();
     }
 
     /**
@@ -619,7 +627,7 @@ public static File[] getCurrentDirectories(File[] dirs) {
 
     public Bookie(ServerConfiguration conf)
             throws IOException, InterruptedException, BookieException {
-        this(conf, NullStatsLogger.INSTANCE, PooledByteBufAllocator.DEFAULT, BookieServiceInfo.NO_INFO);
+        this(conf, NullStatsLogger.INSTANCE, PooledByteBufAllocator.DEFAULT, new SimpleBookieServiceInfoProvider(conf));
     }
 
     private static LedgerStorage buildLedgerStorage(ServerConfiguration conf) throws IOException {

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java
@@ -68,15 +68,15 @@ public class Cookie {
 
     static final int CURRENT_COOKIE_LAYOUT_VERSION = 4;
     private final int layoutVersion;
-    private final String bookieHost;
+    private final String bookieId;
     private final String journalDirs;
     private final String ledgerDirs;
     private final String instanceId;
     private static final String SEPARATOR = "\t";
 
-    private Cookie(int layoutVersion, String bookieHost, String journalDirs, String ledgerDirs, String instanceId) {
+    private Cookie(int layoutVersion, String bookieId, String journalDirs, String ledgerDirs, String instanceId) {
         this.layoutVersion = layoutVersion;
-        this.bookieHost = bookieHost;
+        this.bookieId = bookieId;
         this.journalDirs = journalDirs;
         this.ledgerDirs = ledgerDirs;
         this.instanceId = instanceId;
@@ -130,7 +130,7 @@ private void verifyInternal(Cookie c, boolean checkIfSuperSet) throws BookieExce
             errMsg = "Cookie is of too old version " + c.layoutVersion;
             LOG.error(errMsg);
             throw new BookieException.InvalidCookieException(errMsg);
-        } else if (!(c.layoutVersion >= 3 && c.bookieHost.equals(bookieHost)
+        } else if (!(c.layoutVersion >= 3 && c.bookieId.equals(bookieId)
             && c.journalDirs.equals(journalDirs) && verifyLedgerDirs(c, checkIfSuperSet))) {
             errMsg = "Cookie [" + this + "] is not matching with [" + c + "]";
             throw new BookieException.InvalidCookieException(errMsg);
@@ -157,7 +157,7 @@ public String toString() {
             return toStringVersion3();
         }
         CookieFormat.Builder builder = CookieFormat.newBuilder();
-        builder.setBookieHost(bookieHost);
+        builder.setBookieHost(bookieId);
         builder.setJournalDir(journalDirs);
         builder.setLedgerDirs(ledgerDirs);
         if (null != instanceId) {
@@ -172,7 +172,7 @@ public String toString() {
     private String toStringVersion3() {
         StringBuilder b = new StringBuilder();
         b.append(CURRENT_COOKIE_LAYOUT_VERSION).append("\n")
-            .append(bookieHost).append("\n")
+            .append(bookieId).append("\n")
             .append(journalDirs).append("\n")
             .append(ledgerDirs).append("\n");
         return b.toString();
@@ -193,14 +193,14 @@ private static Builder parse(BufferedReader reader) throws IOException {
                     + "', cannot parse cookie.");
         }
         if (layoutVersion == 3) {
-            cBuilder.setBookieHost(reader.readLine());
+            cBuilder.setBookieId(reader.readLine());
             cBuilder.setJournalDirs(reader.readLine());
             cBuilder.setLedgerDirs(reader.readLine());
         } else if (layoutVersion >= 4) {
             CookieFormat.Builder cfBuilder = CookieFormat.newBuilder();
             TextFormat.merge(reader, cfBuilder);
             CookieFormat data = cfBuilder.build();
-            cBuilder.setBookieHost(data.getBookieHost());
+            cBuilder.setBookieId(data.getBookieHost());
             cBuilder.setJournalDirs(data.getJournalDir());
             cBuilder.setLedgerDirs(data.getLedgerDirs());
             // Since InstanceId is optional
@@ -293,7 +293,7 @@ static Builder generateCookie(ServerConfiguration conf)
             throws UnknownHostException {
         Builder builder = Cookie.newBuilder();
         builder.setLayoutVersion(CURRENT_COOKIE_LAYOUT_VERSION);
-        builder.setBookieHost(Bookie.getBookieId(conf).toString());
+        builder.setBookieId(Bookie.getBookieId(conf).toString());
         builder.setJournalDirs(Joiner.on(',').join(conf.getJournalDirNames()));
         builder.setLedgerDirs(encodeDirPaths(conf.getLedgerDirNames()));
         return builder;
@@ -364,15 +364,17 @@ public static Cookie readFromDirectory(File directory) throws IOException {
      *         if the 'bookieHost' was created using a hostname
      */
     public boolean isBookieHostCreatedFromIp() throws IOException {
-        String[] parts = bookieHost.split(":");
+        String[] parts = bookieId.split(":");
         if (parts.length != 2) {
-            throw new IOException(bookieHost + " does not have the form host:port");
+            // custom BookieId ?
+            return false;
         }
         int port;
         try {
             port = Integer.parseInt(parts[1]);
         } catch (NumberFormatException e) {
-            throw new IOException(bookieHost + " does not have the form host:port");
+            // custom BookieId ?
+            return false;
         }
 
         InetSocketAddress addr = new InetSocketAddress(parts[0], port);
@@ -384,18 +386,18 @@ public boolean isBookieHostCreatedFromIp() throws IOException {
      */
     public static class Builder {
         private int layoutVersion = CURRENT_COOKIE_LAYOUT_VERSION;
-        private String bookieHost = null;
+        private String bookieId = null;
         private String journalDirs = null;
         private String ledgerDirs = null;
         private String instanceId = null;
 
         private Builder() {
         }
 
-        private Builder(int layoutVersion, String bookieHost, String journalDirs, String ledgerDirs,
+        private Builder(int layoutVersion, String bookieId, String journalDirs, String ledgerDirs,
                         String instanceId) {
             this.layoutVersion = layoutVersion;
-            this.bookieHost = bookieHost;
+            this.bookieId = bookieId;
             this.journalDirs = journalDirs;
             this.ledgerDirs = ledgerDirs;
             this.instanceId = instanceId;
@@ -406,8 +408,8 @@ public Builder setLayoutVersion(int layoutVersion) {
             return this;
         }
 
-        public Builder setBookieHost(String bookieHost) {
-            this.bookieHost = bookieHost;
+        public Builder setBookieId(String bookieId) {
+            this.bookieId = bookieId;
             return this;
         }
 
@@ -427,7 +429,7 @@ public Builder setInstanceId(String instanceId) {
         }
 
         public Cookie build() {
-            return new Cookie(layoutVersion, bookieHost, journalDirs, ledgerDirs, instanceId);
+            return new Cookie(layoutVersion, bookieId, journalDirs, ledgerDirs, instanceId);
         }
     }
 
@@ -447,7 +449,7 @@ public static Builder newBuilder() {
      * @return cookie builder
      */
     public static Builder newBuilder(Cookie oldCookie) {
-        return new Builder(oldCookie.layoutVersion, oldCookie.bookieHost, oldCookie.journalDirs, oldCookie.ledgerDirs,
+        return new Builder(oldCookie.layoutVersion, oldCookie.bookieId, oldCookie.journalDirs, oldCookie.ledgerDirs,
                 oldCookie.instanceId);
     }
 }
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java
@@ -38,6 +38,7 @@
 import org.apache.bookkeeper.common.util.ReflectionUtils;
 import org.apache.bookkeeper.discover.RegistrationManager;
 import org.apache.bookkeeper.discover.ZKRegistrationManager;
+import org.apache.bookkeeper.net.BookieId;
 import org.apache.bookkeeper.stats.NullStatsProvider;
 import org.apache.bookkeeper.stats.StatsProvider;
 import org.apache.commons.configuration.ConfigurationException;
@@ -148,6 +149,7 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
     protected static final String LISTENING_INTERFACE = "listeningInterface";
     protected static final String ALLOW_LOOPBACK = "allowLoopback";
     protected static final String ADVERTISED_ADDRESS = "advertisedAddress";
+    protected static final String BOOKIE_ID = "bookieId";
     protected static final String ALLOW_EPHEMERAL_PORTS = "allowEphemeralPorts";
 
     protected static final String JOURNAL_DIR = "journalDirectory";
@@ -1006,6 +1008,38 @@ public ServerConfiguration setAllowLoopback(boolean allow) {
         return this;
     }
 
+     /**
+     * Get the configured BookieId for the bookie.
+     *
+     * <p>If present, this setting will take precedence over the
+     * automatic BookieId generation, based on Network Addresses.
+     *
+     * @see #setBookieId(java.lang.String)
+     * @see #getAdvertisedAddress()
+     * @return the configure address to be advertised
+     */
+    public String getBookieId() {
+        return this.getString(BOOKIE_ID, null);
+    }
+
+    /**
+     * Configure the bookie to advertise a specific BookieId.
+     *
+     * <p>By default, a bookie will advertise a BookieId computed
+     * from the primary network endpoint addresss.
+     *
+     * @see #getBookieId()
+     * @see #setAdvertisedAddress(java.lang.String)
+     * @param bookieId the bookie id
+     *
+     * @return server configuration
+     */
+    public ServerConfiguration setBookieId(String bookieId) {
+        BookieId.parse(bookieId);
+        this.setProperty(BOOKIE_ID, bookieId);
+        return this;
+    }
+
     /**
      * Get the configured advertised address for the bookie.
      *

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
@@ -258,7 +258,9 @@ private CompletableFuture<Versioned<BookieServiceInfo>> readBookieServiceInfo(Bo
                     bookieServiceInfoCache.put(bookieId, result);
                     promise.complete(result);
                 } catch (IOException ex) {
-                    promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc), path));
+                    log.error("Cannot update BookieInfo for {}", ex);
+                    promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc), path)
+                            .initCause(ex));
                     return;
                 }
             } else if (KeeperException.Code.NONODE.intValue() == rc) {
@@ -273,7 +275,9 @@ private CompletableFuture<Versioned<BookieServiceInfo>> readBookieServiceInfo(Bo
                             bookieServiceInfoCache.put(bookieId, result);
                             promise.complete(result);
                         } catch (IOException ex) {
-                            promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc2), path2));
+                            log.error("Cannot update BookieInfo for {}", ex);
+                            promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc2), path2)
+                                    .initCause(ex));
                             return;
                         }
                     } else {
@@ -343,14 +347,18 @@ private CompletableFuture<Versioned<Set<BookieId>>> getChildren(String regPath,
                     bookieInfoUpdated.add(readBookieServiceInfo(id));
                 }
             }
-            FutureUtils
-                    .collect(bookieInfoUpdated)
-                    .whenComplete((List<Versioned<BookieServiceInfo>> info, Throwable error) -> {
-                        // we are ignoring errors intentionally
-                        // there could be bookies that published unparseable information
-                        // or other temporary/permanent or temporary errors
-                        future.complete(new Versioned<>(bookies, version));
-                    });
+            if (bookieInfoUpdated.isEmpty()) {
+                future.complete(new Versioned<>(bookies, version));
+            } else {
+                FutureUtils
+                        .collect(bookieInfoUpdated)
+                        .whenComplete((List<Versioned<BookieServiceInfo>> info, Throwable error) -> {
+                            // we are ignoring errors intentionally
+                            // there could be bookies that publish unparseable information
+                            // or other temporary/permanent errors
+                            future.complete(new Versioned<>(bookies, version));
+                        });
+            }
         }, null);
         return future;
     }

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/net/BookieSocketAddress.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/net/BookieSocketAddress.java
@@ -24,7 +24,6 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.google.common.net.InetAddresses;
-import io.netty.channel.local.LocalAddress;
 
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
@@ -106,15 +105,6 @@ public InetSocketAddress getSocketAddress() {
         });
     }
 
-    /**
-     * Maps the socketAddress to a "local" address.
-     */
-    @JsonIgnore
-    public LocalAddress getLocalAddress() {
-        // for local address, we just need "port" to differentiate different addresses.
-        return new LocalAddress("" + port);
-    }
-
     // Return the String "serialized" version of this object.
     @Override
     public String toString() {

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java
@@ -44,6 +44,7 @@
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.ChannelGroupFuture;
 import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.channel.local.LocalAddress;
 import io.netty.channel.local.LocalChannel;
 import io.netty.channel.local.LocalServerChannel;
 import io.netty.channel.socket.SocketChannel;
@@ -77,6 +78,7 @@
 import org.apache.bookkeeper.common.collections.BlockingMpscQueue;
 import org.apache.bookkeeper.common.util.affinity.CpuAffinity;
 import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.net.BookieId;
 import org.apache.bookkeeper.net.BookieSocketAddress;
 import org.apache.bookkeeper.processor.RequestProcessor;
 import org.apache.bookkeeper.util.ByteBufList;
@@ -103,6 +105,7 @@ class BookieNettyServer {
     volatile boolean suspended = false;
     ChannelGroup allChannels;
     final BookieSocketAddress bookieAddress;
+    final BookieId bookieId;
     final InetSocketAddress bindAddress;
 
     final BookieAuthProvider.Factory authProviderFactory;
@@ -160,7 +163,7 @@ protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
         } else {
             jvmEventLoopGroup = null;
         }
-
+        bookieId = Bookie.getBookieId(conf);
         bookieAddress = Bookie.getBookieAddress(conf);
         if (conf.getListeningInterface() == null) {
             bindAddress = new InetSocketAddress(conf.getBookiePort());
@@ -410,10 +413,10 @@ protected void initChannel(LocalChannel ch) throws Exception {
                     pipeline.addLast("contextHandler", contextHandler);
                 }
             });
-
+            LOG.info("Binding jvm bookie-rpc endpoint to {}", bookieId.toString());
             // use the same address 'name', so clients can find local Bookie still discovering them using ZK
-            jvmBootstrap.bind(bookieAddress.getLocalAddress()).sync();
-            LocalBookiesRegistry.registerLocalBookieAddress(bookieAddress.toBookieId());
+            jvmBootstrap.bind(new LocalAddress(bookieId.toString())).sync();
+            LocalBookiesRegistry.registerLocalBookieAddress(bookieId);
         }
     }
 

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java
@@ -44,6 +44,7 @@
 import org.apache.bookkeeper.common.util.JsonUtil.ParseJsonException;
 import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.discover.BookieServiceInfoUtils;
 import org.apache.bookkeeper.net.BookieId;
 import org.apache.bookkeeper.net.BookieSocketAddress;
 import org.apache.bookkeeper.processor.RequestProcessor;
@@ -85,13 +86,23 @@ public class BookieServer {
     public BookieServer(ServerConfiguration conf) throws IOException,
             KeeperException, InterruptedException, BookieException,
             UnavailableException, CompatibilityException, SecurityException {
-        this(conf, NullStatsLogger.INSTANCE, BookieServiceInfo.NO_INFO);
+        this(conf, NullStatsLogger.INSTANCE, null);
     }
 
     public BookieServer(ServerConfiguration conf, StatsLogger statsLogger,
             Supplier<BookieServiceInfo> bookieServiceInfoProvider)
             throws IOException, KeeperException, InterruptedException,
             BookieException, UnavailableException, CompatibilityException, SecurityException {
+        if (bookieServiceInfoProvider == null) {
+            bookieServiceInfoProvider = () -> {
+                try {
+                    return BookieServiceInfoUtils
+                            .buildLegacyBookieServiceInfo(this.getLocalAddress().toBookieId().toString());
+                } catch (IOException err) {
+                    throw new RuntimeException(err);
+                }
+            };
+        }
         this.conf = conf;
         validateUser(conf);
         String configAsString;