You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by eo...@apache.org on 2020/05/27 06:14:13 UTC
[bookkeeper] branch master updated: BP-38: Publish Bookie Service
Info on Metadata Service
This is an automated email from the ASF dual-hosted git repository.
eolivelli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 637bfa7 BP-38: Publish Bookie Service Info on Metadata Service
637bfa7 is described below
commit 637bfa756ddba08bf5f67fc2d45124afd5dbc0bc
Author: Matteo Minardi <mi...@hotmail.it>
AuthorDate: Wed May 27 08:14:01 2020 +0200
BP-38: Publish Bookie Service Info on Metadata Service
Starting from the implementation of the #2213 PR
### Changes
- Adding the ability for the Bookie to advertise on the Metadata Service (ZooKeeper) the exposed ports/services, like the HTTP service, Metrics
- Implemented the Protobuf serialization of the BookieServiceInfo
- Added the command EndpointInfoCommand to retrieve remote bookie endpoint information both for "bkctl" and "bookkeeper shell".
Usage: `
bin/bookkeeper shell endpointinfo -bookieid <bookieaddress>
`
Master Issue: #2215
Proposal Doc: https://bookkeeper.apache.org/bps/BP-38-bookie-endpoint-discovery
Reviewers: Jia Zhai <zh...@apache.org>, Enrico Olivelli <eo...@gmail.com>
This closes #2314 from mino181295/single/endpoint-discovery-impl
---
.../common/component/ComponentInfoPublisher.java | 146 ++++++++++++++++++
.../common/component/ComponentStarter.java | 2 +
.../common/component/LifecycleComponent.java | 3 +
.../common/component/LifecycleComponentStack.java | 31 ++++
.../common/component/TestComponentStarter.java | 1 +
bookkeeper-proto/src/main/proto/DataFormats.proto | 23 +++
.../java/org/apache/bookkeeper/bookie/Bookie.java | 35 ++---
.../org/apache/bookkeeper/bookie/BookieShell.java | 46 ++++++
.../bookkeeper/bookie/BookieStateManager.java | 18 ++-
.../apache/bookkeeper/bookie/ReadOnlyBookie.java | 10 +-
.../apache/bookkeeper/client/BookKeeperAdmin.java | 10 ++
.../bookkeeper/discover/BookieServiceInfo.java | 166 +++++++++++++++++++++
.../discover/BookieServiceInfoUtils.java | 51 +++++++
.../bookkeeper/discover/RegistrationClient.java | 21 +++
.../bookkeeper/discover/RegistrationManager.java | 3 +-
.../bookkeeper/discover/ZKRegistrationClient.java | 73 +++++++++
.../bookkeeper/discover/ZKRegistrationManager.java | 54 +++++--
.../org/apache/bookkeeper/proto/BookieServer.java | 16 +-
.../java/org/apache/bookkeeper/server/Main.java | 41 ++++-
.../bookkeeper/server/service/BookieService.java | 35 ++++-
.../bookkeeper/server/service/HttpService.java | 14 ++
.../cli/commands/bookies/EndpointInfoCommand.java | 125 ++++++++++++++++
.../apache/bookkeeper/util/LocalBookKeeper.java | 74 +++++----
.../bookie/BookieInitializationTest.java | 61 +++++++-
.../apache/bookkeeper/bookie/StateManagerTest.java | 3 +-
.../bookkeeper/client/BookKeeperAdminTest.java | 79 ++++++++++
.../bookkeeper/discover/BookieServiceInfoTest.java | 91 +++++++++++
.../AuditorPlacementPolicyCheckTest.java | 11 +-
.../replication/AuditorReplicasCheckTest.java | 3 +-
.../org/apache/bookkeeper/server/TestMain.java | 18 ++-
.../bookkeeper/test/BookKeeperClusterTestCase.java | 11 +-
.../metadata/etcd/EtcdRegistrationManager.java | 4 +-
.../metadata/etcd/EtcdRegistrationTest.java | 19 +--
.../etcd/testing/EtcdBKClusterTestBase.java | 4 +-
site/_data/cli/shell.yaml | 2 +
.../resolver/BKRegistrationNameResolverTest.java | 5 +-
.../java/org/apache/distributedlog/TestTxnId.java | 4 +-
.../bookkeeper/stream/server/StorageServer.java | 35 ++++-
.../stream/server/service/BookieService.java | 10 +-
.../server/service/RegistrationStateService.java | 4 +-
.../tools/cli/commands/BookieCommandGroup.java | 2 +
41 files changed, 1240 insertions(+), 124 deletions(-)
diff --git a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/ComponentInfoPublisher.java b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/ComponentInfoPublisher.java
new file mode 100644
index 0000000..0085e82
--- /dev/null
+++ b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/ComponentInfoPublisher.java
@@ -0,0 +1,146 @@
+/*
+ * 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.common.component;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Allows a component to publish information about
+ * the services it implements, the endpoints it exposes
+ * and other useful information for management tools and client.
+ */
+@Slf4j
+public class ComponentInfoPublisher {
+
+ private final Map<String, String> properties = new ConcurrentHashMap<>();
+ private final Map<String, EndpointInfo> endpoints = new ConcurrentHashMap<>();
+
+ /**
+ * Endpoint information.
+ */
+ public static final class EndpointInfo {
+
+ private final String id;
+ private final int port;
+ private final String host;
+ private final String protocol;
+ private final List<String> auth;
+ private final List<String> extensions;
+
+ public EndpointInfo(String id, int port, String host, String protocol,
+ List<String> auth, List<String> extensions) {
+ this.id = id;
+ this.port = port;
+ this.host = host;
+ this.protocol = protocol;
+ this.auth = auth == null ? Collections.emptyList() : Collections.unmodifiableList(auth);
+ this.extensions = extensions == null ? Collections.emptyList() : Collections.unmodifiableList(extensions);
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public String getProtocol() {
+ return protocol;
+ }
+
+ public List<String> getAuth() {
+ return auth;
+ }
+
+ public List<String> getExtensions() {
+ return extensions;
+ }
+
+ @Override
+ public String toString() {
+ return "EndpointInfo{" + "id=" + id + ", port=" + port + ", host=" + host + ", protocol=" + protocol + ", "
+ + "auth=" + auth + ", extensions=" + extensions + '}';
+ }
+
+ }
+
+ private volatile boolean startupFinished;
+
+ /**
+ * Publish an information about the system, like an endpoint address.
+ *
+ * @param key the key
+ * @param value the value, null values are not allowed.
+ */
+ public void publishProperty(String key, String value) {
+ if (log.isDebugEnabled()) {
+ log.debug("publish {}={}", key, value);
+ }
+ if (startupFinished) {
+ throw new IllegalStateException("Server already started, cannot publish " + key);
+ }
+ Objects.requireNonNull(key);
+ Objects.requireNonNull(value, "Value for " + key + " cannot be null");
+
+ properties.put(key, value);
+ }
+
+ public void publishEndpoint(EndpointInfo endpoint) {
+ if (log.isDebugEnabled()) {
+ log.debug("publishEndpoint {} on {}", endpoint, this);
+ }
+ EndpointInfo exists = endpoints.put(endpoint.id, endpoint);
+ if (exists != null) {
+ throw new IllegalStateException("An endpoint with id " + endpoint.id
+ + " has already been published: " + exists);
+ }
+ }
+
+ public Map<String, String> getProperties() {
+ if (!startupFinished) {
+ throw new IllegalStateException("Startup not yet finished");
+ }
+ return Collections.unmodifiableMap(properties);
+ }
+
+ public Map<String, EndpointInfo> getEndpoints() {
+ if (!startupFinished) {
+ throw new IllegalStateException("Startup not yet finished");
+ }
+ return Collections.unmodifiableMap(endpoints);
+ }
+
+ /**
+ * Called by the framework to signal that preparation of startup is done,
+ * so we have gathered all of the available information.
+ */
+ public void startupFinished() {
+ startupFinished = true;
+ }
+
+}
diff --git a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/ComponentStarter.java b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/ComponentStarter.java
index acb643b..50fd81a 100644
--- a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/ComponentStarter.java
+++ b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/ComponentStarter.java
@@ -79,6 +79,8 @@ public class ComponentStarter {
shutdownHookThread.start();
});
+ component.publishInfo(new ComponentInfoPublisher());
+
log.info("Starting component {}.", component.getName());
component.start();
log.info("Started component {}.", component.getName());
diff --git a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/LifecycleComponent.java b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/LifecycleComponent.java
index d382066..836cffb 100644
--- a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/LifecycleComponent.java
+++ b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/LifecycleComponent.java
@@ -33,6 +33,9 @@ public interface LifecycleComponent extends AutoCloseable {
void removeLifecycleListener(LifecycleListener listener);
+ default void publishInfo(ComponentInfoPublisher componentInfoPublisher) {
+ }
+
void start();
void stop();
diff --git a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/LifecycleComponentStack.java b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/LifecycleComponentStack.java
index d606910..5b94e04 100644
--- a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/LifecycleComponentStack.java
+++ b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/LifecycleComponentStack.java
@@ -26,10 +26,12 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import java.lang.Thread.UncaughtExceptionHandler;
import java.util.List;
+import lombok.extern.slf4j.Slf4j;
/**
* A stack of {@link LifecycleComponent}s.
*/
+@Slf4j
public class LifecycleComponentStack implements LifecycleComponent {
public static Builder newBuilder() {
@@ -42,12 +44,19 @@ public class LifecycleComponentStack implements LifecycleComponent {
public static class Builder {
private String name;
+ private ComponentInfoPublisher componentInfoPublisher;
private final List<LifecycleComponent> components;
private Builder() {
components = Lists.newArrayList();
}
+ public Builder withComponentInfoPublisher(ComponentInfoPublisher componentInfoPublisher) {
+ checkNotNull(componentInfoPublisher, "ComponentInfoPublisher is null");
+ this.componentInfoPublisher = componentInfoPublisher;
+ return this;
+ }
+
public Builder addComponent(LifecycleComponent component) {
checkNotNull(component, "Lifecycle component is null");
components.add(component);
@@ -64,6 +73,7 @@ public class LifecycleComponentStack implements LifecycleComponent {
checkArgument(!components.isEmpty(), "Lifecycle component stack is empty : " + components);
return new LifecycleComponentStack(
name,
+ componentInfoPublisher != null ? componentInfoPublisher : new ComponentInfoPublisher(),
ImmutableList.copyOf(components));
}
@@ -71,10 +81,13 @@ public class LifecycleComponentStack implements LifecycleComponent {
private final String name;
private final ImmutableList<LifecycleComponent> components;
+ private final ComponentInfoPublisher componentInfoPublisher;
private LifecycleComponentStack(String name,
+ ComponentInfoPublisher componentInfoPublisher,
ImmutableList<LifecycleComponent> components) {
this.name = name;
+ this.componentInfoPublisher = componentInfoPublisher;
this.components = components;
}
@@ -109,7 +122,25 @@ public class LifecycleComponentStack implements LifecycleComponent {
}
@Override
+ public void publishInfo(ComponentInfoPublisher componentInfoPublisher) {
+ components.forEach(component -> {
+ if (log.isDebugEnabled()) {
+ log.debug("calling publishInfo on {} ", component);
+ }
+ component.publishInfo(componentInfoPublisher);
+ });
+ }
+
+ @Override
public void start() {
+ components.forEach(component -> {
+ if (log.isDebugEnabled()) {
+ log.debug("calling publishInfo on {} ", component);
+ }
+ component.publishInfo(componentInfoPublisher);
+ });
+ componentInfoPublisher.startupFinished();
+
components.forEach(component -> component.start());
}
diff --git a/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/component/TestComponentStarter.java b/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/component/TestComponentStarter.java
index ed97adc..ef35a9d 100644
--- a/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/component/TestComponentStarter.java
+++ b/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/component/TestComponentStarter.java
@@ -42,6 +42,7 @@ public class TestComponentStarter {
LifecycleComponent component = mock(LifecycleComponent.class);
when(component.getName()).thenReturn("test-start-component");
ComponentStarter.startComponent(component);
+ verify(component).publishInfo(any(ComponentInfoPublisher.class));
verify(component).start();
}
diff --git a/bookkeeper-proto/src/main/proto/DataFormats.proto b/bookkeeper-proto/src/main/proto/DataFormats.proto
index 6591d65..0af1bb1 100644
--- a/bookkeeper-proto/src/main/proto/DataFormats.proto
+++ b/bookkeeper-proto/src/main/proto/DataFormats.proto
@@ -118,3 +118,26 @@ message PlacementPolicyCheckFormat {
message ReplicasCheckFormat {
optional int64 replicasCheckCTime = 1;
}
+
+/**
+ * information about services exposed by a Bookie.
+ */
+message BookieServiceInfoFormat {
+
+ /**
+ * Information about an endpoint.
+ */
+ message Endpoint {
+ required string id = 1;
+ required int32 port = 2;
+ required string host = 3;
+ required string protocol = 4;
+
+ repeated string auth = 5;
+ repeated string extensions = 6;
+ }
+
+ repeated Endpoint endpoints = 6;
+ map<string, string> properties = 7;
+}
+
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
index 7a655d6..6a2209a 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
@@ -57,6 +57,7 @@ import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.apache.bookkeeper.bookie.BookieException.BookieIllegalOpException;
@@ -73,6 +74,7 @@ import org.apache.bookkeeper.bookie.stats.BookieStats;
import org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorage;
import org.apache.bookkeeper.common.util.Watcher;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
@@ -123,6 +125,7 @@ public class Bookie extends BookieCriticalThread {
static final long METAENTRY_ID_LEDGER_EXPLICITLAC = -0x8000;
private final LedgerDirsManager ledgerDirsManager;
+ protected final Supplier<BookieServiceInfo> bookieServiceInfoProvider;
private final LedgerDirsManager indexDirsManager;
LedgerDirsMonitor dirsMonitor;
@@ -611,7 +614,7 @@ public class Bookie extends BookieCriticalThread {
public Bookie(ServerConfiguration conf)
throws IOException, InterruptedException, BookieException {
- this(conf, NullStatsLogger.INSTANCE, PooledByteBufAllocator.DEFAULT);
+ this(conf, NullStatsLogger.INSTANCE, PooledByteBufAllocator.DEFAULT, BookieServiceInfo.NO_INFO);
}
private static LedgerStorage buildLedgerStorage(ServerConfiguration conf) throws IOException {
@@ -674,9 +677,11 @@ public class Bookie extends BookieCriticalThread {
return ledgerStorage;
}
- public Bookie(ServerConfiguration conf, StatsLogger statsLogger, ByteBufAllocator allocator)
+ public Bookie(ServerConfiguration conf, StatsLogger statsLogger,
+ ByteBufAllocator allocator, Supplier<BookieServiceInfo> bookieServiceInfoProvider)
throws IOException, InterruptedException, BookieException {
super("Bookie-" + conf.getBookiePort());
+ this.bookieServiceInfoProvider = bookieServiceInfoProvider;
this.statsLogger = statsLogger;
this.conf = conf;
this.journalDirectories = Lists.newArrayList();
@@ -797,7 +802,8 @@ public class Bookie extends BookieCriticalThread {
}
StateManager initializeStateManager() throws IOException {
- return new BookieStateManager(conf, statsLogger, metadataDriver, ledgerDirsManager);
+ return new BookieStateManager(conf, statsLogger, metadataDriver,
+ ledgerDirsManager, bookieServiceInfoProvider);
}
void readJournal() throws IOException, BookieException {
@@ -1584,29 +1590,6 @@ public class Bookie extends BookieCriticalThread {
}
/**
- * @param args
- * @throws IOException
- * @throws InterruptedException
- */
- public static void main(String[] args)
- throws IOException, InterruptedException, BookieException {
- Bookie b = new Bookie(new ServerConfiguration());
- b.start();
- CounterCallback cb = new CounterCallback();
- long start = System.currentTimeMillis();
- for (int i = 0; i < 100000; i++) {
- ByteBuf buff = Unpooled.buffer(1024);
- buff.writeLong(1);
- buff.writeLong(i);
- cb.incCount();
- b.addEntry(buff, false /* ackBeforeSync */, cb, null, new byte[0]);
- }
- cb.waitZero();
- long end = System.currentTimeMillis();
- System.out.println("Took " + (end - start) + "ms");
- }
-
- /**
* Returns exit code - cause of failure.
*
* @return {@link ExitCode}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java
index bf4b57c..00c99a7 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java
@@ -64,6 +64,7 @@ import org.apache.bookkeeper.tools.cli.commands.bookie.RebuildDBLedgerLocationsI
import org.apache.bookkeeper.tools.cli.commands.bookie.RegenerateInterleavedStorageIndexFileCommand;
import org.apache.bookkeeper.tools.cli.commands.bookie.SanityTestCommand;
import org.apache.bookkeeper.tools.cli.commands.bookies.DecommissionCommand;
+import org.apache.bookkeeper.tools.cli.commands.bookies.EndpointInfoCommand;
import org.apache.bookkeeper.tools.cli.commands.bookies.InfoCommand;
import org.apache.bookkeeper.tools.cli.commands.bookies.InstanceIdCommand;
import org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand;
@@ -137,6 +138,7 @@ public class BookieShell implements Tool {
static final String CMD_DELETELEDGER = "deleteledger";
static final String CMD_BOOKIEINFO = "bookieinfo";
static final String CMD_DECOMMISSIONBOOKIE = "decommissionbookie";
+ static final String CMD_ENDPOINTINFO = "endpointinfo";
static final String CMD_LOSTBOOKIERECOVERYDELAY = "lostbookierecoverydelay";
static final String CMD_TRIGGERAUDIT = "triggeraudit";
static final String CMD_CONVERT_TO_DB_STORAGE = "convert-to-db-storage";
@@ -1708,6 +1710,49 @@ public class BookieShell implements Tool {
}
/**
+ * Command to retrieve remote bookie endpoint information.
+ */
+ class EndpointInfoCmd extends MyCommand {
+ Options lOpts = new Options();
+
+ EndpointInfoCmd() {
+ super(CMD_ENDPOINTINFO);
+ lOpts.addOption("b", "bookieid", true, "Bookie Id");
+ }
+
+ @Override
+ String getDescription() {
+ return "Get info about a remote bookie with a specific bookie address (bookieid)";
+ }
+
+ @Override
+ String getUsage() {
+ return CMD_ENDPOINTINFO + " [-bookieid <bookieaddress>]";
+ }
+
+ @Override
+ Options getOptions() {
+ return lOpts;
+ }
+
+ @Override
+ public int runCmd(CommandLine cmdLine) throws Exception {
+ EndpointInfoCommand cmd = new EndpointInfoCommand();
+ EndpointInfoCommand.EndpointInfoFlags flags = new EndpointInfoCommand.EndpointInfoFlags();
+ final String bookieId = cmdLine.getOptionValue("bookieid");
+ flags.bookie(bookieId);
+ if (StringUtils.isBlank(bookieId)) {
+ LOG.error("Invalid argument list!");
+ this.printUsage();
+ return -1;
+ }
+
+ boolean result = cmd.apply(bkConf, flags);
+ return (result) ? 0 : -1;
+ }
+ }
+
+ /**
* A facility for reporting update ledger progress.
*/
public interface UpdateLedgerNotifier {
@@ -1907,6 +1952,7 @@ public class BookieShell implements Tool {
commands.put(CMD_DELETELEDGER, new DeleteLedgerCmd());
commands.put(CMD_BOOKIEINFO, new BookieInfoCmd());
commands.put(CMD_DECOMMISSIONBOOKIE, new DecommissionBookieCmd());
+ commands.put(CMD_ENDPOINTINFO, new EndpointInfoCmd());
commands.put(CMD_CONVERT_TO_DB_STORAGE, new ConvertToDbStorageCmd());
commands.put(CMD_CONVERT_TO_INTERLEAVED_STORAGE, new ConvertToInterleavedStorageCmd());
commands.put(CMD_REBUILD_DB_LEDGER_LOCATIONS_INDEX, new RebuildDbLedgerLocationsIndexCmd());
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieStateManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieStateManager.java
index 54d6993..45b48fb 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieStateManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieStateManager.java
@@ -39,6 +39,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.MetadataBookieDriver;
import org.apache.bookkeeper.stats.Gauge;
@@ -61,6 +62,7 @@ import org.slf4j.LoggerFactory;
public class BookieStateManager implements StateManager {
private static final Logger LOG = LoggerFactory.getLogger(BookieStateManager.class);
private final ServerConfiguration conf;
+ private final Supplier<BookieServiceInfo> bookieServiceInfoProvider;
private final List<File> statusDirs;
// use an executor to execute the state changes task
@@ -90,7 +92,8 @@ public class BookieStateManager implements StateManager {
public BookieStateManager(ServerConfiguration conf,
StatsLogger statsLogger,
MetadataBookieDriver metadataDriver,
- LedgerDirsManager ledgerDirsManager) throws IOException {
+ LedgerDirsManager ledgerDirsManager,
+ Supplier<BookieServiceInfo> bookieServiceInfoProvider) throws IOException {
this(
conf,
statsLogger,
@@ -102,18 +105,21 @@ public class BookieStateManager implements StateManager {
} catch (UnknownHostException e) {
throw new UncheckedIOException("Failed to resolve bookie id", e);
}
- });
+ },
+ bookieServiceInfoProvider);
}
public BookieStateManager(ServerConfiguration conf,
StatsLogger statsLogger,
Supplier<RegistrationManager> rm,
List<File> statusDirs,
- Supplier<String> bookieIdSupplier) throws IOException {
+ Supplier<String> bookieIdSupplier,
+ Supplier<BookieServiceInfo> bookieServiceInfoProvider) throws IOException {
this.conf = conf;
this.rm = rm;
this.statusDirs = statusDirs;
// ZK ephemeral node for this Bookie.
this.bookieId = bookieIdSupplier.get();
+ this.bookieServiceInfoProvider = bookieServiceInfoProvider;
// 1 : up, 0 : readonly, -1 : unregistered
this.serverStatusGauge = new Gauge<Number>() {
@Override
@@ -143,7 +149,7 @@ public class BookieStateManager implements StateManager {
BookieStateManager(ServerConfiguration conf, MetadataBookieDriver metadataDriver) throws IOException {
this(conf, NullStatsLogger.INSTANCE, metadataDriver, new LedgerDirsManager(conf, conf.getLedgerDirs(),
new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()),
- NullStatsLogger.INSTANCE));
+ NullStatsLogger.INSTANCE), BookieServiceInfo.NO_INFO);
}
@Override
@@ -263,7 +269,7 @@ public class BookieStateManager implements StateManager {
rmRegistered.set(false);
try {
- rm.get().registerBookie(bookieId, isReadOnly);
+ rm.get().registerBookie(bookieId, isReadOnly, bookieServiceInfoProvider.get());
rmRegistered.set(true);
} catch (BookieException e) {
throw new IOException(e);
@@ -333,7 +339,7 @@ public class BookieStateManager implements StateManager {
return;
}
try {
- rm.get().registerBookie(bookieId, true);
+ rm.get().registerBookie(bookieId, true, bookieServiceInfoProvider.get());
} catch (BookieException e) {
LOG.error("Error in transition to ReadOnly Mode."
+ " Shutting down", e);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/ReadOnlyBookie.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/ReadOnlyBookie.java
index cb6e9f0..3c84a06 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/ReadOnlyBookie.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/ReadOnlyBookie.java
@@ -24,7 +24,9 @@ package org.apache.bookkeeper.bookie;
import io.netty.buffer.ByteBufAllocator;
import java.io.IOException;
+import java.util.function.Supplier;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
@@ -40,9 +42,10 @@ public class ReadOnlyBookie extends Bookie {
private static final Logger LOG = LoggerFactory.getLogger(ReadOnlyBookie.class);
- public ReadOnlyBookie(ServerConfiguration conf, StatsLogger statsLogger, ByteBufAllocator allocator)
+ public ReadOnlyBookie(ServerConfiguration conf, StatsLogger statsLogger,
+ ByteBufAllocator allocator, Supplier<BookieServiceInfo> bookieServiceInfoProvider)
throws IOException, KeeperException, InterruptedException, BookieException {
- super(conf, statsLogger, allocator);
+ super(conf, statsLogger, allocator, bookieServiceInfoProvider);
if (conf.isReadOnlyModeEnabled()) {
stateManager.forceToReadOnly();
} else {
@@ -55,7 +58,8 @@ public class ReadOnlyBookie extends Bookie {
@Override
StateManager initializeStateManager() throws IOException {
- return new BookieStateManager(conf, statsLogger, metadataDriver, getLedgerDirsManager()) {
+ return new BookieStateManager(conf, statsLogger, metadataDriver, getLedgerDirsManager(),
+ bookieServiceInfoProvider) {
@Override
public void doTransitionToWritableMode() {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
index cac1d9d..ea6cd9f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
@@ -52,6 +52,7 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.Predicate;
+import lombok.SneakyThrows;
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieException;
@@ -62,8 +63,10 @@ import org.apache.bookkeeper.client.LedgerFragmentReplicator.SingleFragmentCallb
import org.apache.bookkeeper.client.SyncCallbackUtils.SyncOpenCallback;
import org.apache.bookkeeper.client.SyncCallbackUtils.SyncReadCallback;
import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.RegistrationClient.RegistrationListener;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManager.LedgerRangeIterator;
@@ -226,6 +229,13 @@ public class BookKeeperAdmin implements AutoCloseable {
return bkc.bookieWatcher.getAllBookies();
}
+ @SneakyThrows
+ public BookieServiceInfo getBookieServiceInfo(String bookiedId)
+ throws BKException {
+ return FutureUtils.result(bkc.getMetadataClientDriver()
+ .getRegistrationClient().getBookieServiceInfo(bookiedId)).getValue();
+ }
+
/**
* Get a list of readonly bookies synchronously.
*
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/BookieServiceInfo.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/BookieServiceInfo.java
new file mode 100644
index 0000000..8b23a70
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/BookieServiceInfo.java
@@ -0,0 +1,166 @@
+/*
+ * 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;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/**
+ * Information about services exposed by a Bookie.
+ */
+public final class BookieServiceInfo {
+
+ /**
+ * Default empty implementation.
+ */
+ public static final BookieServiceInfo EMPTY = new BookieServiceInfo(
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+
+ /**
+ * Default empty implementation.
+ */
+ public static final Supplier<BookieServiceInfo> NO_INFO = () -> EMPTY;
+
+ private Map<String, String> properties;
+ private List<Endpoint> endpoints;
+
+ public BookieServiceInfo(Map<String, String> properties, List<Endpoint> endpoints) {
+ this.properties = Collections.unmodifiableMap(properties);
+ this.endpoints = Collections.unmodifiableList(endpoints);
+ }
+
+ public BookieServiceInfo() {
+ this(Collections.emptyMap(), Collections.emptyList());
+ }
+
+ /**
+ * Unmodifiable map with bookie wide information.
+ *
+ * @return the map
+ */
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+
+ /**
+ * Unmodifieable structure with the list of exposed endpoints.
+ *
+ * @return the list.
+ */
+ public List<Endpoint> getEndpoints() {
+ return endpoints;
+ }
+
+ public void setProperties(Map<String, String> properties) {
+ this.properties = properties;
+ }
+
+ public void setEndpoints(List<Endpoint> endpoints) {
+ this.endpoints = endpoints;
+ }
+
+ /**
+ * Information about an endpoint.
+ */
+ public static final class Endpoint {
+
+ private String id;
+ private int port;
+ private String host;
+ private String protocol;
+ private List<String> auth;
+ private List<String> extensions;
+
+ public Endpoint(String id, int port, String host, String protocol, List<String> auth, List<String> extensions) {
+ this.id = id;
+ this.port = port;
+ this.host = host;
+ this.protocol = protocol;
+ this.auth = auth;
+ this.extensions = extensions;
+ }
+
+ public Endpoint() {
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ public String getProtocol() {
+ return protocol;
+ }
+
+ public void setProtocol(String protocol) {
+ this.protocol = protocol;
+ }
+
+ public List<String> getAuth() {
+ return auth;
+ }
+
+ public void setAuth(List<String> auth) {
+ this.auth = auth;
+ }
+
+ public List<String> getExtensions() {
+ return extensions;
+ }
+
+ public void setExtensions(List<String> extensions) {
+ this.extensions = extensions;
+ }
+
+ @Override
+ public String toString() {
+ return "EndpointInfo{" + "id=" + id + ", port=" + port + ", host=" + host + ", protocol=" + protocol + ", "
+ + "auth=" + auth + ", extensions=" + extensions + '}';
+ }
+
+ }
+
+ @Override
+ public String toString() {
+ return "BookieServiceInfo{" + "properties=" + properties + ", endpoints=" + endpoints + '}';
+ }
+
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/BookieServiceInfoUtils.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/BookieServiceInfoUtils.java
new file mode 100644
index 0000000..252cb3b
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/BookieServiceInfoUtils.java
@@ -0,0 +1,51 @@
+/*
+ * 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;
+
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.Collections;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+
+/**
+ * Utility class for {@link BookieServiceInfo}.
+ */
+public final class BookieServiceInfoUtils {
+
+ /**
+ * Creates a default legacy bookie info implementation.
+ * In the default implementation there is one endpoint with
+ * <code>bookie-rpc</code> protocol and the bookie id in the host port.
+ *
+ * @param bookieId bookie id
+ * @return default implementation of a BookieServiceInfo
+ * @throws UnknownHostException if the given bookieId is invalid
+ */
+ public static BookieServiceInfo buildLegacyBookieServiceInfo(String bookieId) throws UnknownHostException {
+ BookieSocketAddress address = new BookieSocketAddress(bookieId);
+ BookieServiceInfo.Endpoint endpoint = new BookieServiceInfo.Endpoint();
+ endpoint.setId(bookieId);
+ endpoint.setHost(address.getHostName());
+ endpoint.setPort(address.getPort());
+ endpoint.setProtocol("bookie-rpc");
+ endpoint.setAuth(Collections.emptyList());
+ endpoint.setExtensions(Collections.emptyList());
+ return new BookieServiceInfo(Collections.emptyMap(), Arrays.asList(endpoint));
+ }
+
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationClient.java
index 4056329..f7823c0 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationClient.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationClient.java
@@ -18,11 +18,14 @@
package org.apache.bookkeeper.discover;
+import java.net.UnknownHostException;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.common.annotation.InterfaceAudience.LimitedPrivate;
import org.apache.bookkeeper.common.annotation.InterfaceStability.Evolving;
+import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Versioned;
/**
@@ -66,6 +69,24 @@ public interface RegistrationClient extends AutoCloseable {
CompletableFuture<Versioned<Set<BookieSocketAddress>>> getReadOnlyBookies();
/**
+ * Get detailed information about the services exposed by a Bookie.
+ * For old bookies it is expected to return an empty BookieServiceInfo structure.
+ *
+ * @param bookieId this is the id of the bookie, it can be computed from a {@link BookieSocketAddress}
+ * @return a future represents the available information.
+ *
+ * @since 4.11
+ */
+ default CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(String bookieId) {
+ try {
+ BookieServiceInfo bookieServiceInfo = BookieServiceInfoUtils.buildLegacyBookieServiceInfo(bookieId);
+ return FutureUtils.value(new Versioned<>(bookieServiceInfo, new LongVersion(-1)));
+ } catch (UnknownHostException e) {
+ return FutureUtils.exception(e);
+ }
+ }
+
+ /**
* Watch the changes of bookies.
*
* <p>The topology changes of bookies will be propagated to the provided <i>listener</i>.
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationManager.java
index 3d357d4..e5b0f6a 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationManager.java
@@ -59,9 +59,10 @@ public interface RegistrationManager extends AutoCloseable {
*
* @param bookieId bookie id
* @param readOnly whether to register it as writable or readonly
+ * @param serviceInfo information about services exposed by the Bookie
* @throws BookieException when fail to register a bookie.
*/
- void registerBookie(String bookieId, boolean readOnly) throws BookieException;
+ void registerBookie(String bookieId, boolean readOnly, BookieServiceInfo serviceInfo) throws BookieException;
/**
* Unregistering the bookie server as <i>bookieId</i>.
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
index 6629f0c..6995a14 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
@@ -22,6 +22,7 @@ import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
import static org.apache.bookkeeper.util.BookKeeperConstants.COOKIE_NODE;
import static org.apache.bookkeeper.util.BookKeeperConstants.READONLY;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import java.io.IOException;
import java.util.HashSet;
@@ -33,23 +34,28 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BKException.Code;
import org.apache.bookkeeper.client.BKException.ZKException;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.common.util.SafeRunnable;
import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.proto.DataFormats.BookieServiceInfoFormat;
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Version;
import org.apache.bookkeeper.versioning.Version.Occurred;
import org.apache.bookkeeper.versioning.Versioned;
+import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
/**
* ZooKeeper based {@link RegistrationClient}.
@@ -212,6 +218,73 @@ public class ZKRegistrationClient implements RegistrationClient {
return getChildren(bookieReadonlyRegistrationPath, null);
}
+ @Override
+ public CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(String bookieId) {
+ String pathAsWritable = bookieRegistrationPath + "/" + bookieId;
+ String pathAsReadonly = bookieReadonlyRegistrationPath + "/" + bookieId;
+
+ CompletableFuture<Versioned<BookieServiceInfo>> promise = new CompletableFuture<>();
+ zk.getData(pathAsWritable, false, (int rc, String path, Object o, byte[] bytes, Stat stat) -> {
+ if (KeeperException.Code.OK.intValue() == rc) {
+ try {
+ BookieServiceInfo bookieServiceInfo = deserializeBookieServiceInfo(bookieId, bytes);
+ promise.complete(new Versioned<>(bookieServiceInfo, new LongVersion(stat.getCversion())));
+ } catch (IOException ex) {
+ promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc), path));
+ return;
+ }
+ } else if (KeeperException.Code.NONODE.intValue() == rc) {
+ // not found, looking for a readonly bookie
+ zk.getData(pathAsReadonly, false, (int rc2, String path2, Object o2, byte[] bytes2, Stat stat2) -> {
+ if (KeeperException.Code.OK.intValue() == rc2) {
+ try {
+ BookieServiceInfo bookieServiceInfo = deserializeBookieServiceInfo(bookieId, bytes2);
+ promise.complete(new Versioned<>(bookieServiceInfo, new LongVersion(stat2.getCversion())));
+ } catch (IOException ex) {
+ promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc2), path2));
+ return;
+ }
+ } else {
+ // not found as writable and readonly, the bookie is offline
+ promise.completeExceptionally(BKException.create(BKException.Code.NoBookieAvailableException));
+ }
+ }, null);
+ } else {
+ promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc), path));
+ }
+ }, null);
+ return promise;
+ }
+
+ @SuppressWarnings("unchecked")
+ @VisibleForTesting
+ static BookieServiceInfo deserializeBookieServiceInfo(String bookieId, byte[] bookieServiceInfo)
+ throws IOException {
+ if (bookieServiceInfo == null || bookieServiceInfo.length == 0) {
+ return BookieServiceInfoUtils.buildLegacyBookieServiceInfo(bookieId);
+ }
+
+ BookieServiceInfoFormat builder = BookieServiceInfoFormat.parseFrom(bookieServiceInfo);
+ BookieServiceInfo bsi = new BookieServiceInfo();
+ List<BookieServiceInfo.Endpoint> endpoints = builder.getEndpointsList().stream()
+ .map(e -> {
+ BookieServiceInfo.Endpoint endpoint = new BookieServiceInfo.Endpoint();
+ endpoint.setId(e.getId());
+ endpoint.setPort(e.getPort());
+ endpoint.setHost(e.getHost());
+ endpoint.setProtocol(e.getProtocol());
+ endpoint.setAuth(e.getAuthList());
+ endpoint.setExtensions(e.getExtensionsList());
+ return endpoint;
+ })
+ .collect(Collectors.toList());
+
+ bsi.setEndpoints(endpoints);
+ bsi.setProperties(builder.getPropertiesMap());
+
+ return bsi;
+ }
+
private CompletableFuture<Versioned<Set<BookieSocketAddress>>> getChildren(String regPath, Watcher watcher) {
CompletableFuture<Versioned<Set<BookieSocketAddress>>> future = FutureUtils.createFuture();
zk.getChildren(regPath, watcher, (rc, path, ctx, children, stat) -> {
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 e9bd784..9272eb8 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
@@ -18,15 +18,17 @@
package org.apache.bookkeeper.discover;
-import static java.nio.charset.StandardCharsets.UTF_8;
+import static com.google.common.base.Charsets.UTF_8;
import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
import static org.apache.bookkeeper.util.BookKeeperConstants.COOKIE_NODE;
import static org.apache.bookkeeper.util.BookKeeperConstants.EMPTY_BYTE_ARRAY;
import static org.apache.bookkeeper.util.BookKeeperConstants.INSTANCEID;
import static org.apache.bookkeeper.util.BookKeeperConstants.READONLY;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
@@ -34,6 +36,7 @@ import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
+import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.bookie.BookieException.BookieIllegalOpException;
@@ -52,6 +55,7 @@ import org.apache.bookkeeper.meta.ZkLayoutManager;
import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.proto.DataFormats.BookieServiceInfoFormat;
import org.apache.bookkeeper.util.BookKeeperConstants;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.bookkeeper.versioning.LongVersion;
@@ -213,21 +217,53 @@ public class ZKRegistrationManager implements RegistrationManager {
}
@Override
- public void registerBookie(String bookieId, boolean readOnly) throws BookieException {
+ public void registerBookie(String bookieId, boolean readOnly,
+ BookieServiceInfo bookieServiceInfo) throws BookieException {
if (!readOnly) {
String regPath = bookieRegistrationPath + "/" + bookieId;
- doRegisterBookie(regPath);
+ doRegisterBookie(regPath, bookieServiceInfo);
} else {
- doRegisterReadOnlyBookie(bookieId);
+ doRegisterReadOnlyBookie(bookieId, bookieServiceInfo);
}
}
- private void doRegisterBookie(String regPath) throws BookieException {
+ @VisibleForTesting
+ static byte[] serializeBookieServiceInfo(BookieServiceInfo bookieServiceInfo) {
+ if (log.isDebugEnabled()) {
+ log.debug("serialize BookieServiceInfo {}", bookieServiceInfo);
+ }
+ try (ByteArrayOutputStream os = new ByteArrayOutputStream()) {
+ BookieServiceInfoFormat.Builder builder = BookieServiceInfoFormat.newBuilder();
+ List<BookieServiceInfoFormat.Endpoint> bsiEndpoints = bookieServiceInfo.getEndpoints().stream()
+ .map(e -> {
+ return BookieServiceInfoFormat.Endpoint.newBuilder()
+ .setId(e.getId())
+ .setPort(e.getPort())
+ .setHost(e.getHost())
+ .setProtocol(e.getProtocol())
+ .addAllAuth(e.getAuth())
+ .addAllExtensions(e.getExtensions())
+ .build();
+ })
+ .collect(Collectors.toList());
+
+ builder.addAllEndpoints(bsiEndpoints);
+ builder.putAllProperties(bookieServiceInfo.getProperties());
+
+ builder.build().writeTo(os);
+ return os.toByteArray();
+ } catch (IOException err) {
+ log.error("Cannot serialize bookieServiceInfo from " + bookieServiceInfo);
+ throw new RuntimeException(err);
+ }
+ }
+
+ private void doRegisterBookie(String regPath, BookieServiceInfo bookieServiceInfo) throws BookieException {
// ZK ephemeral node for this Bookie.
try {
if (!checkRegNodeAndWaitExpired(regPath)) {
// Create the ZK ephemeral node for this Bookie.
- zk.create(regPath, new byte[0], zkAcls, CreateMode.EPHEMERAL);
+ zk.create(regPath, serializeBookieServiceInfo(bookieServiceInfo), zkAcls, CreateMode.EPHEMERAL);
zkRegManagerInitialized = true;
}
} catch (KeeperException ke) {
@@ -248,11 +284,11 @@ public class ZKRegistrationManager implements RegistrationManager {
}
}
- private void doRegisterReadOnlyBookie(String bookieId) throws BookieException {
+ private void doRegisterReadOnlyBookie(String bookieId, BookieServiceInfo bookieServiceInfo) throws BookieException {
try {
if (null == zk.exists(this.bookieReadonlyRegistrationPath, false)) {
try {
- zk.create(this.bookieReadonlyRegistrationPath, new byte[0],
+ zk.create(this.bookieReadonlyRegistrationPath, serializeBookieServiceInfo(bookieServiceInfo),
zkAcls, CreateMode.PERSISTENT);
} catch (NodeExistsException e) {
// this node is just now created by someone.
@@ -260,7 +296,7 @@ public class ZKRegistrationManager implements RegistrationManager {
}
String regPath = bookieReadonlyRegistrationPath + "/" + bookieId;
- doRegisterBookie(regPath);
+ doRegisterBookie(regPath, bookieServiceInfo);
// clear the write state
regPath = bookieRegistrationPath + "/" + bookieId;
try {
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
index 14d6393..fda7ed7 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java
@@ -33,6 +33,7 @@ import java.net.UnknownHostException;
import java.security.AccessControlException;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieCriticalThread;
@@ -42,6 +43,7 @@ import org.apache.bookkeeper.bookie.ReadOnlyBookie;
import org.apache.bookkeeper.common.allocator.ByteBufAllocatorBuilder;
import org.apache.bookkeeper.common.util.JsonUtil.ParseJsonException;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.processor.RequestProcessor;
import org.apache.bookkeeper.replication.ReplicationException.CompatibilityException;
@@ -82,10 +84,11 @@ public class BookieServer {
public BookieServer(ServerConfiguration conf) throws IOException,
KeeperException, InterruptedException, BookieException,
UnavailableException, CompatibilityException, SecurityException {
- this(conf, NullStatsLogger.INSTANCE);
+ this(conf, NullStatsLogger.INSTANCE, BookieServiceInfo.NO_INFO);
}
- public BookieServer(ServerConfiguration conf, StatsLogger statsLogger)
+ public BookieServer(ServerConfiguration conf, StatsLogger statsLogger,
+ Supplier<BookieServiceInfo> bookieServiceInfoProvider)
throws IOException, KeeperException, InterruptedException,
BookieException, UnavailableException, CompatibilityException, SecurityException {
this.conf = conf;
@@ -102,7 +105,7 @@ public class BookieServer {
this.statsLogger = statsLogger;
this.nettyServer = new BookieNettyServer(this.conf, null, allocator);
try {
- this.bookie = newBookie(conf, allocator);
+ this.bookie = newBookie(conf, allocator, bookieServiceInfoProvider);
} catch (IOException | KeeperException | InterruptedException | BookieException e) {
// interrupted on constructing a bookie
this.nettyServer.shutdown();
@@ -129,11 +132,12 @@ public class BookieServer {
this.uncaughtExceptionHandler = exceptionHandler;
}
- protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator)
+ protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator,
+ Supplier<BookieServiceInfo> bookieServiceInfoProvider)
throws IOException, KeeperException, InterruptedException, BookieException {
return conf.isForceReadOnlyBookie()
- ? new ReadOnlyBookie(conf, statsLogger.scope(BOOKIE_SCOPE), allocator)
- : new Bookie(conf, statsLogger.scope(BOOKIE_SCOPE), allocator);
+ ? new ReadOnlyBookie(conf, statsLogger.scope(BOOKIE_SCOPE), allocator, bookieServiceInfoProvider)
+ : new Bookie(conf, statsLogger.scope(BOOKIE_SCOPE), allocator, bookieServiceInfoProvider);
}
public void start() throws InterruptedException {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/Main.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/Main.java
index ec163f8..b569d1e 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/Main.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/Main.java
@@ -26,14 +26,19 @@ import java.net.MalformedURLException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.bookie.ExitCode;
import org.apache.bookkeeper.bookie.ScrubberStats;
+import org.apache.bookkeeper.common.component.ComponentInfoPublisher;
import org.apache.bookkeeper.common.component.ComponentStarter;
import org.apache.bookkeeper.common.component.LifecycleComponent;
import org.apache.bookkeeper.common.component.LifecycleComponentStack;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.UncheckedConfigurationException;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.discover.BookieServiceInfo.Endpoint;
import org.apache.bookkeeper.server.component.ServerLifecycleComponent;
import org.apache.bookkeeper.server.conf.BookieConfiguration;
import org.apache.bookkeeper.server.http.BKHttpServiceProvider;
@@ -287,19 +292,26 @@ public class Main {
* @return lifecycle stack
*/
public static LifecycleComponentStack buildBookieServer(BookieConfiguration conf) throws Exception {
- LifecycleComponentStack.Builder serverBuilder = LifecycleComponentStack.newBuilder().withName("bookie-server");
+
+ final ComponentInfoPublisher componentInfoPublisher = new ComponentInfoPublisher();
+
+ final Supplier<BookieServiceInfo> bookieServiceInfoProvider =
+ () -> buildBookieServiceInfo(componentInfoPublisher);
+ LifecycleComponentStack.Builder serverBuilder = LifecycleComponentStack
+ .newBuilder()
+ .withComponentInfoPublisher(componentInfoPublisher)
+ .withName("bookie-server");
// 1. build stats provider
StatsProviderService statsProviderService =
new StatsProviderService(conf);
StatsLogger rootStatsLogger = statsProviderService.getStatsProvider().getStatsLogger("");
-
serverBuilder.addComponent(statsProviderService);
log.info("Load lifecycle component : {}", StatsProviderService.class.getName());
// 2. build bookie server
BookieService bookieService =
- new BookieService(conf, rootStatsLogger);
+ new BookieService(conf, rootStatsLogger, bookieServiceInfoProvider);
serverBuilder.addComponent(bookieService);
log.info("Load lifecycle component : {}", BookieService.class.getName());
@@ -329,7 +341,6 @@ public class Main {
.build();
HttpService httpService =
new HttpService(provider, conf, rootStatsLogger);
-
serverBuilder.addComponent(httpService);
log.info("Load lifecycle component : {}", HttpService.class.getName());
}
@@ -359,4 +370,26 @@ public class Main {
return serverBuilder.build();
}
+ /**
+ * Create the {@link BookieServiceInfo} starting from the published endpoints.
+ *
+ * @see ComponentInfoPublisher
+ * @param componentInfoPublisher the endpoint publisher
+ * @return the created bookie service info
+ */
+ private static BookieServiceInfo buildBookieServiceInfo(ComponentInfoPublisher componentInfoPublisher) {
+ List<Endpoint> endpoints = componentInfoPublisher.getEndpoints().values()
+ .stream().map(e -> {
+ return new Endpoint(
+ e.getId(),
+ e.getPort(),
+ e.getHost(),
+ e.getProtocol(),
+ e.getAuth(),
+ e.getExtensions()
+ );
+ }).collect(Collectors.toList());
+ return new BookieServiceInfo(componentInfoPublisher.getProperties(), endpoints);
+ }
+
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/service/BookieService.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/service/BookieService.java
index 645da03..530dd0c 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/service/BookieService.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/service/BookieService.java
@@ -20,6 +20,15 @@ package org.apache.bookkeeper.server.service;
import java.io.IOException;
import java.lang.Thread.UncaughtExceptionHandler;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.common.component.ComponentInfoPublisher;
+import org.apache.bookkeeper.common.component.ComponentInfoPublisher.EndpointInfo;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.server.component.ServerLifecycleComponent;
import org.apache.bookkeeper.server.conf.BookieConfiguration;
@@ -28,6 +37,7 @@ import org.apache.bookkeeper.stats.StatsLogger;
/**
* A {@link ServerLifecycleComponent} that starts the core bookie server.
*/
+@Slf4j
public class BookieService extends ServerLifecycleComponent {
public static final String NAME = "bookie-server";
@@ -35,10 +45,11 @@ public class BookieService extends ServerLifecycleComponent {
private final BookieServer server;
public BookieService(BookieConfiguration conf,
- StatsLogger statsLogger)
+ StatsLogger statsLogger,
+ Supplier<BookieServiceInfo> bookieServiceInfoProvider)
throws Exception {
super(NAME, conf, statsLogger);
- this.server = new BookieServer(conf.getServerConf(), statsLogger);
+ this.server = new BookieServer(conf.getServerConf(), statsLogger, bookieServiceInfoProvider);
}
@Override
@@ -69,4 +80,24 @@ public class BookieService extends ServerLifecycleComponent {
protected void doClose() throws IOException {
this.server.shutdown();
}
+
+ @Override
+ public void publishInfo(ComponentInfoPublisher componentInfoPublisher) {
+ try {
+ BookieSocketAddress localAddress = getServer().getLocalAddress();
+ List<String> extensions = new ArrayList<>();
+ if (conf.getServerConf().getTLSProviderFactoryClass() != null) {
+ extensions.add("tls");
+ }
+ EndpointInfo endpoint = new EndpointInfo("bookie",
+ localAddress.getPort(),
+ localAddress.getHostName(),
+ "bookie-rpc", null, extensions);
+ componentInfoPublisher.publishEndpoint(endpoint);
+
+ } catch (UnknownHostException err) {
+ log.error("Cannot compute local address", err);
+ }
+ }
+
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/service/HttpService.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/service/HttpService.java
index 8a86ffa..e5dffc9 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/service/HttpService.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/service/HttpService.java
@@ -21,7 +21,9 @@ package org.apache.bookkeeper.server.service;
import static com.google.common.base.Preconditions.checkNotNull;
import java.io.IOException;
+import org.apache.bookkeeper.common.component.ComponentInfoPublisher;
+import org.apache.bookkeeper.common.component.ComponentInfoPublisher.EndpointInfo;
import org.apache.bookkeeper.http.HttpServer;
import org.apache.bookkeeper.http.HttpServerLoader;
import org.apache.bookkeeper.server.component.ServerLifecycleComponent;
@@ -63,4 +65,16 @@ public class HttpService extends ServerLifecycleComponent {
protected void doClose() throws IOException {
server.stopServer();
}
+
+ @Override
+ public void publishInfo(ComponentInfoPublisher componentInfoPublisher) {
+ if (conf.getServerConf().isHttpServerEnabled()) {
+ EndpointInfo endpoint = new EndpointInfo("httpserver",
+ conf.getServerConf().getHttpServerPort(),
+ "0.0.0.0",
+ "http", null, null);
+ componentInfoPublisher.publishEndpoint(endpoint);
+ }
+ }
+
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/EndpointInfoCommand.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/EndpointInfoCommand.java
new file mode 100644
index 0000000..3ef7e7b
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/EndpointInfoCommand.java
@@ -0,0 +1,125 @@
+/*
+ * 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.tools.cli.commands.bookies;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import java.io.IOException;
+import java.util.Collection;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.tools.cli.helpers.BookieCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Get endpoint information about a Bookie.
+ */
+public class EndpointInfoCommand extends BookieCommand<EndpointInfoCommand.EndpointInfoFlags> {
+
+ static final Logger LOG = LoggerFactory.getLogger(EndpointInfoCommand.class);
+
+ private static final String NAME = "endpointinfo";
+ private static final String DESC = "Get all end point information about a given bookie.";
+
+ public EndpointInfoCommand() {
+ this(new EndpointInfoFlags());
+ }
+
+ private EndpointInfoCommand(EndpointInfoFlags flags) {
+ super(CliSpec.<EndpointInfoFlags>newBuilder().withName(NAME).withDescription(DESC).withFlags(flags).build());
+ }
+
+ /**
+ * Flags for this command.
+ */
+ @Accessors(fluent = true)
+ @Setter
+ public static class EndpointInfoFlags extends CliFlags {
+
+ @Parameter(required = true, names = {"-b", "--bookieid"}, description = "Get information about a remote bookie")
+ private String bookie;
+
+ }
+
+ @Override
+ public boolean apply(ServerConfiguration conf, EndpointInfoFlags cmdFlags) {
+ try {
+ return getEndpointInfo(conf, cmdFlags);
+ } catch (Exception e) {
+ throw new UncheckedExecutionException(e.getMessage(), e);
+ }
+ }
+
+ private boolean getEndpointInfo(ServerConfiguration conf, EndpointInfoFlags flags)
+ throws BKException, InterruptedException, IOException {
+ ClientConfiguration adminConf = new ClientConfiguration(conf);
+ BookKeeperAdmin admin = new BookKeeperAdmin(adminConf);
+ try {
+ final String bookieId = flags.bookie;
+ if (bookieId == null || bookieId.isEmpty()) {
+ throw new IllegalArgumentException("BookieId is required");
+ }
+ BookieSocketAddress address = new BookieSocketAddress(bookieId);
+ Collection<BookieSocketAddress> allBookies = admin.getAllBookies();
+ if (!allBookies.contains(address)) {
+ System.out.println("Bookie " + bookieId + " does not exist, only " + allBookies);
+ return false;
+ }
+ BookieServiceInfo bookieServiceInfo = admin.getBookieServiceInfo(bookieId);
+
+ System.out.println("BookiedId: " + bookieId);
+ if (!bookieServiceInfo.getProperties().isEmpty()) {
+ System.out.println("Properties");
+ bookieServiceInfo.getProperties().forEach((k, v) -> {
+ System.out.println(k + ":" + v);
+ });
+ }
+ if (!bookieServiceInfo.getEndpoints().isEmpty()) {
+ bookieServiceInfo.getEndpoints().forEach(e -> {
+ System.out.println("Endpoint: " + e.getId());
+ System.out.println("Protocol: " + e.getProtocol());
+ System.out.println("Address: " + e.getHost() + ":" + e.getPort());
+ System.out.println("Auth: " + e.getAuth());
+ System.out.println("Extensions: " + e.getExtensions());
+ });
+ } else {
+ System.out.println("Bookie did not publish any endpoint info. Maybe it is down");
+ return false;
+ }
+
+ return true;
+ } catch (Exception e) {
+ LOG.error("Received exception in EndpointInfoCommand ", e);
+ return false;
+ } finally {
+ if (admin != null) {
+ admin.close();
+ }
+ }
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/LocalBookKeeper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/LocalBookKeeper.java
index b078dd4..0bfc923 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/LocalBookKeeper.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/LocalBookKeeper.java
@@ -34,19 +34,21 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
import org.apache.bookkeeper.bookie.Bookie;
-import org.apache.bookkeeper.bookie.BookieException;
-import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.common.component.ComponentInfoPublisher;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.discover.BookieServiceInfo.Endpoint;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.proto.BookieServer;
-import org.apache.bookkeeper.replication.ReplicationException.CompatibilityException;
-import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
+import org.apache.bookkeeper.server.conf.BookieConfiguration;
+import org.apache.bookkeeper.server.service.BookieService;
import org.apache.bookkeeper.shims.zk.ZooKeeperServerShim;
import org.apache.bookkeeper.shims.zk.ZooKeeperServerShimFactory;
-import org.apache.bookkeeper.tls.SecurityException;
+import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
-import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.io.FileUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
@@ -164,39 +166,20 @@ public class LocalBookKeeper {
}
private List<File> runBookies(String dirSuffix)
- throws IOException, KeeperException, InterruptedException, BookieException,
- UnavailableException, CompatibilityException, SecurityException, BKException,
- ConfigurationException {
+ throws Exception {
List<File> tempDirs = new ArrayList<File>();
try {
runBookies(tempDirs, dirSuffix);
return tempDirs;
- } catch (IOException ioe) {
+ } catch (Exception ioe) {
cleanupDirectories(tempDirs);
throw ioe;
- } catch (KeeperException ke) {
- cleanupDirectories(tempDirs);
- throw ke;
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- cleanupDirectories(tempDirs);
- throw ie;
- } catch (BookieException be) {
- cleanupDirectories(tempDirs);
- throw be;
- } catch (UnavailableException ue) {
- cleanupDirectories(tempDirs);
- throw ue;
- } catch (CompatibilityException ce) {
- cleanupDirectories(tempDirs);
- throw ce;
}
}
@SuppressWarnings("deprecation")
private void runBookies(List<File> tempDirs, String dirSuffix)
- throws IOException, KeeperException, InterruptedException, BookieException, UnavailableException,
- CompatibilityException, SecurityException, BKException, ConfigurationException {
+ throws Exception {
LOG.info("Starting Bookie(s)");
// Create Bookie Servers (B1, B2, B3)
@@ -272,8 +255,18 @@ public class LocalBookKeeper {
String fileName = Bookie.getBookieAddress(bsConfs[i]).toString() + ".conf";
serializeLocalBookieConfig(bsConfs[i], fileName);
- bs[i] = new BookieServer(bsConfs[i]);
- bs[i].start();
+ // Mimic BookKeeper Main
+ final ComponentInfoPublisher componentInfoPublisher = new ComponentInfoPublisher();
+ final Supplier<BookieServiceInfo> bookieServiceInfoProvider =
+ () -> buildBookieServiceInfo(componentInfoPublisher);
+ BookieService bookieService = new BookieService(new BookieConfiguration(bsConfs[i]),
+ NullStatsLogger.INSTANCE,
+ bookieServiceInfoProvider
+ );
+ bs[i] = bookieService.getServer();
+ bookieService.publishInfo(componentInfoPublisher);
+ componentInfoPublisher.startupFinished();
+ bookieService.start();
}
/*
@@ -547,4 +540,25 @@ public class LocalBookKeeper {
}
}
+ /**
+ * Create the {@link BookieServiceInfo} starting from the published endpoints.
+ *
+ * @see ComponentInfoPublisher
+ * @param componentInfoPublisher the endpoint publisher
+ * @return the created bookie service info
+ */
+ private static BookieServiceInfo buildBookieServiceInfo(ComponentInfoPublisher componentInfoPublisher) {
+ List<Endpoint> endpoints = componentInfoPublisher.getEndpoints().values()
+ .stream().map(e -> {
+ return new Endpoint(
+ e.getId(),
+ e.getPort(),
+ e.getHost(),
+ e.getProtocol(),
+ e.getAuth(),
+ e.getExtensions()
+ );
+ }).collect(Collectors.toList());
+ return new BookieServiceInfo(componentInfoPublisher.getProperties(), endpoints);
+ }
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
index bff9e55..2fcc8a7 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
@@ -35,6 +35,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doThrow;
@@ -59,6 +60,7 @@ import java.net.URL;
import java.net.URLConnection;
import java.security.AccessControlException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -67,6 +69,7 @@ import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
import org.apache.bookkeeper.bookie.BookieException.DiskPartitionDuplicationException;
import org.apache.bookkeeper.bookie.BookieException.MetadataStoreException;
@@ -83,6 +86,8 @@ import org.apache.bookkeeper.common.component.LifecycleComponent;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.discover.BookieServiceInfo.Endpoint;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.http.HttpRouter;
import org.apache.bookkeeper.http.HttpServerLoader;
@@ -92,6 +97,7 @@ import org.apache.bookkeeper.meta.zk.ZKMetadataBookieDriver;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookieServer;
+import org.apache.bookkeeper.proto.DataFormats.BookieServiceInfoFormat;
import org.apache.bookkeeper.replication.AutoRecoveryMain;
import org.apache.bookkeeper.replication.ReplicationException.CompatibilityException;
import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
@@ -293,12 +299,13 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
RegistrationManager rm = mock(RegistrationManager.class);
doThrow(new MetadataStoreException("mocked exception"))
.when(rm)
- .registerBookie(anyString(), anyBoolean());
+ .registerBookie(anyString(), anyBoolean(), any(BookieServiceInfo.class));
// simulating ZooKeeper exception by assigning a closed zk client to bk
BookieServer bkServer = new BookieServer(conf) {
@Override
- protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator)
+ protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator,
+ Supplier<BookieServiceInfo> bookieServiceInfoProvider)
throws IOException, KeeperException, InterruptedException,
BookieException {
Bookie bookie = new Bookie(conf);
@@ -495,6 +502,49 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
}
}
+ @Test(timeout = 20000)
+ public void testBookieRegistrationBookieServiceInfo() throws Exception {
+ final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
+ .setMetadataServiceUri(metadataServiceUri)
+ .setUseHostNameAsBookieID(true)
+ .setUseShortHostName(true)
+ .setListeningInterface(null);
+
+ final String bookieId = InetAddress.getLocalHost().getCanonicalHostName().split("\\.", 2)[0]
+ + ":" + conf.getBookiePort();
+ String bkRegPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(conf) + "/" + AVAILABLE_NODE + "/" + bookieId;
+
+ driver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
+
+ Endpoint endpoint = new Endpoint("test", 1281, "localhost", "bookie-rpc",
+ Collections.emptyList(), Collections.emptyList());
+ BookieServiceInfo bsi = new BookieServiceInfo(Collections.emptyMap(), Arrays.asList(endpoint));
+ Supplier<BookieServiceInfo> supplier = () -> bsi;
+
+ DiskChecker diskChecker = new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold());
+ LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(
+ conf, conf.getLedgerDirs(), diskChecker);
+ try (StateManager manager = new BookieStateManager(conf,
+ NullStatsLogger.INSTANCE, driver, ledgerDirsManager, supplier)) {
+ manager.registerBookie(true).get();
+ assertTrue("Bookie registration node doesn't exists!",
+ driver.getRegistrationManager().isBookieRegistered(bookieId));
+ }
+ Stat bkRegNode = zkc.exists(bkRegPath, false);
+ assertNotNull("Bookie registration has been failed", bkRegNode);
+
+ byte[] bkRegNodeData = zkc.getData(bkRegPath, null, null);
+ assertFalse("Bookie service info not written", bkRegNodeData == null || bkRegNodeData.length == 0);
+
+ BookieServiceInfoFormat serializedBookieServiceInfo = BookieServiceInfoFormat.parseFrom(bkRegNodeData);
+ BookieServiceInfoFormat.Endpoint serializedEndpoint = serializedBookieServiceInfo.getEndpoints(0);
+ assertNotNull("Serialized Bookie endpoint not found", serializedEndpoint);
+
+ assertEquals(endpoint.getId(), serializedEndpoint.getId());
+ assertEquals(endpoint.getHost(), serializedEndpoint.getHost());
+ assertEquals(endpoint.getPort(), serializedEndpoint.getPort());
+ }
+
/**
* Verify user cannot start if user is in permittedStartupUsers conf list BKException BKUnauthorizedAccessException
* if cannot start.
@@ -648,7 +698,7 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
.setMetadataServiceUri(metadataServiceUri);
BookieConfiguration bkConf = new BookieConfiguration(conf);
- BookieService service = new BookieService(bkConf, NullStatsLogger.INSTANCE);
+ BookieService service = new BookieService(bkConf, NullStatsLogger.INSTANCE, BookieServiceInfo.NO_INFO);
CompletableFuture<Void> startFuture = ComponentStarter.startComponent(service);
// shutdown the bookie service
@@ -971,7 +1021,8 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
}
@Override
- protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator)
+ protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator,
+ Supplier<BookieServiceInfo> bookieServiceInfoProvider)
throws IOException, KeeperException, InterruptedException, BookieException {
return new MockBookieWithNoopShutdown(conf, NullStatsLogger.INSTANCE);
}
@@ -980,7 +1031,7 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
class MockBookieWithNoopShutdown extends Bookie {
public MockBookieWithNoopShutdown(ServerConfiguration conf, StatsLogger statsLogger)
throws IOException, KeeperException, InterruptedException, BookieException {
- super(conf, statsLogger, UnpooledByteBufAllocator.DEFAULT);
+ super(conf, statsLogger, UnpooledByteBufAllocator.DEFAULT, BookieServiceInfo.NO_INFO);
}
// making Bookie Shutdown no-op. Ideally for this testcase we need to
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/StateManagerTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/StateManagerTest.java
index 71658e7..ec5b9ef 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/StateManagerTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/StateManagerTest.java
@@ -28,6 +28,7 @@ import io.netty.buffer.UnpooledByteBufAllocator;
import java.io.File;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.meta.MetadataBookieDriver;
import org.apache.bookkeeper.meta.zk.ZKMetadataBookieDriver;
import org.apache.bookkeeper.stats.NullStatsLogger;
@@ -156,7 +157,7 @@ public class StateManagerTest extends BookKeeperClusterTestCase {
.setMetadataServiceUri(zkUtil.getMetadataServiceUri())
.setForceReadOnlyBookie(true);
ReadOnlyBookie readOnlyBookie = new ReadOnlyBookie(readOnlyConf, NullStatsLogger.INSTANCE,
- UnpooledByteBufAllocator.DEFAULT);
+ UnpooledByteBufAllocator.DEFAULT, BookieServiceInfo.NO_INFO);
readOnlyBookie.start();
assertTrue(readOnlyBookie.isRunning());
assertTrue(readOnlyBookie.isReadOnly());
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperAdminTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperAdminTest.java
index daafc74..ecbc4f0 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperAdminTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperAdminTest.java
@@ -23,8 +23,11 @@ package org.apache.bookkeeper.client;
import static com.google.common.base.Charsets.UTF_8;
import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
import static org.apache.bookkeeper.util.BookKeeperConstants.READONLY;
+import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -35,6 +38,7 @@ import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Objects;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -46,15 +50,23 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.common.component.ComponentStarter;
+import org.apache.bookkeeper.common.component.Lifecycle;
+import org.apache.bookkeeper.common.component.LifecycleComponent;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.meta.UnderreplicatedLedger;
import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
+import org.apache.bookkeeper.server.Main;
+import org.apache.bookkeeper.server.conf.BookieConfiguration;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.test.PortManager;
import org.apache.bookkeeper.util.AvailabilityOfEntriesOfLedger;
import org.apache.bookkeeper.util.BookKeeperConstants;
import org.apache.commons.io.FileUtils;
@@ -650,4 +662,71 @@ public class BookKeeperAdminTest extends BookKeeperClusterTestCase {
}
}
}
+
+ private void testBookieServiceInfo(boolean readonly, boolean legacy) throws Exception {
+ File tmpDir = createTempDir("bookie", "test");
+ final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
+ .setJournalDirName(tmpDir.getPath())
+ .setLedgerDirNames(new String[]{tmpDir.getPath()})
+ .setBookiePort(PortManager.nextFreePort())
+ .setMetadataServiceUri(metadataServiceUri);
+
+ LifecycleComponent server = Main.buildBookieServer(new BookieConfiguration(conf));
+ // 2. start the server
+ CompletableFuture<Void> stackComponentFuture = ComponentStarter.startComponent(server);
+ while (server.lifecycleState() != Lifecycle.State.STARTED) {
+ Thread.sleep(100);
+ }
+
+ ServerConfiguration bkConf = newServerConfiguration().setForceReadOnlyBookie(readonly);
+ BookieServer bkServer = startBookie(bkConf);
+
+ String bookieId = bkServer.getLocalAddress().toString();
+ String host = bkServer.getLocalAddress().getHostName();
+ int port = bkServer.getLocalAddress().getPort();
+
+ if (legacy) {
+ String regPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(bkConf) + "/" + AVAILABLE_NODE;
+ regPath = readonly
+ ? regPath + READONLY + "/" + bookieId
+ : regPath + "/" + bookieId;
+ // deleting the metadata, so that the bookie registration should
+ // continue successfully with legacy BookieServiceInfo
+ zkc.setData(regPath, new byte[]{}, -1);
+ }
+
+ try (BookKeeperAdmin bkAdmin = new BookKeeperAdmin(zkUtil.getZooKeeperConnectString())) {
+ BookieServiceInfo bookieServiceInfo = bkAdmin.getBookieServiceInfo(bookieId);
+
+ assertThat(bookieServiceInfo.getEndpoints().size(), is(1));
+ BookieServiceInfo.Endpoint endpoint = bookieServiceInfo.getEndpoints().stream()
+ .filter(e -> Objects.equals(e.getId(), bookieId))
+ .findFirst()
+ .get();
+ assertNotNull("Endpoint " + bookieId + " not found.", endpoint);
+
+ assertThat(endpoint.getHost(), is(host));
+ assertThat(endpoint.getPort(), is(port));
+ assertThat(endpoint.getProtocol(), is("bookie-rpc"));
+ }
+
+ bkServer.shutdown();
+ stackComponentFuture.cancel(true);
+ }
+
+ @Test
+ public void testBookieServiceInfoWritable() throws Exception {
+ testBookieServiceInfo(false, false);
+ }
+
+ @Test
+ public void testBookieServiceInfoReadonly() throws Exception {
+ testBookieServiceInfo(true, false);
+ }
+
+ @Test
+ public void testLegacyBookieServiceInfo() throws Exception {
+ testBookieServiceInfo(false, true);
+ }
+
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/BookieServiceInfoTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/BookieServiceInfoTest.java
new file mode 100644
index 0000000..120a205
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/BookieServiceInfoTest.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2020 The Apache Software Foundation.
+ *
+ * Licensed 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;
+
+import static org.apache.bookkeeper.discover.ZKRegistrationClient.deserializeBookieServiceInfo;
+import static org.apache.bookkeeper.discover.ZKRegistrationManager.serializeBookieServiceInfo;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.bookkeeper.discover.BookieServiceInfo.Endpoint;
+import org.junit.Test;
+
+/**
+ * Unit test of the {@link BookieServiceInfo} serialization/deserialization methods.
+ */
+public class BookieServiceInfoTest {
+
+ @Test
+ public void testSerializeDeserializeBookieServiceInfo() throws Exception {
+ String bookieId = "127.0.0.1:3181";
+ {
+ BookieServiceInfo expected = new BookieServiceInfo();
+ Endpoint endpointRPC = new Endpoint("1", 1281, "localhost", "bookie-rpc",
+ Collections.emptyList(), Collections.emptyList());
+ Endpoint endpointHTTP = new Endpoint("2", 1281, "localhost", "bookie-http",
+ Collections.emptyList(), Collections.emptyList());
+ expected.setEndpoints(Arrays.asList(endpointRPC, endpointHTTP));
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put("test", "value");
+ expected.setProperties(properties);
+
+ byte[] serialized = serializeBookieServiceInfo(expected);
+ BookieServiceInfo deserialized = deserializeBookieServiceInfo(bookieId, serialized);
+
+ assertBookieServiceInfoEquals(expected, deserialized);
+ }
+ }
+
+ @Test
+ public void testDeserializeBookieServiceInfo() throws Exception {
+ String bookieId = "127.0.0.1:3181";
+ {
+ BookieServiceInfo expected = BookieServiceInfoUtils.buildLegacyBookieServiceInfo(bookieId);
+ BookieServiceInfo deserialized = deserializeBookieServiceInfo(bookieId, null);
+
+ assertBookieServiceInfoEquals(expected, deserialized);
+ }
+ {
+ BookieServiceInfo expected = BookieServiceInfoUtils.buildLegacyBookieServiceInfo(bookieId);
+ BookieServiceInfo deserialized = deserializeBookieServiceInfo(bookieId, new byte[]{});
+
+ assertBookieServiceInfoEquals(expected, deserialized);
+ }
+ }
+
+ private void assertBookieServiceInfoEquals(BookieServiceInfo expected, BookieServiceInfo provided) {
+ for (Endpoint ep : expected.getEndpoints()) {
+ Endpoint e = provided.getEndpoints().stream()
+ .filter(ee -> Objects.equals(ee.getId(), ep.getId()))
+ .findFirst()
+ .get();
+ assertThat(e.getHost(), is(ep.getHost()));
+ assertThat(e.getPort(), is(ep.getPort()));
+ assertThat(e.getProtocol(), is(ep.getProtocol()));
+ assertArrayEquals(e.getAuth().toArray(), ep.getAuth().toArray());
+ assertArrayEquals(e.getExtensions().toArray(), ep.getExtensions().toArray());
+ }
+ assertEquals(expected.getProperties(), provided.getProperties());
+ }
+
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTest.java
index 468a032..75a7b16 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTest.java
@@ -42,6 +42,7 @@ import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.client.api.LedgerMetadata;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
@@ -108,7 +109,7 @@ public class AuditorPlacementPolicyCheckTest extends BookKeeperClusterTestCase {
bookieAddress = new BookieSocketAddress("98.98.98." + i, 2181);
StaticDNSResolver.addNodeToRack(bookieAddress.getHostName(), "/rack" + (i));
bookieAddresses.add(bookieAddress);
- regManager.registerBookie(bookieAddress.toString(), false);
+ regManager.registerBookie(bookieAddress.toString(), false, BookieServiceInfo.EMPTY);
}
LedgerManagerFactory mFactory = driver.getLedgerManagerFactory();
@@ -217,7 +218,7 @@ public class AuditorPlacementPolicyCheckTest extends BookKeeperClusterTestCase {
for (int i = 0; i < numOfBookies; i++) {
BookieSocketAddress bookieAddress = new BookieSocketAddress("98.98.98." + i, 2181);
bookieAddresses.add(bookieAddress);
- regManager.registerBookie(bookieAddress.toString(), false);
+ regManager.registerBookie(bookieAddress.toString(), false, BookieServiceInfo.EMPTY);
}
// only three racks
@@ -312,7 +313,7 @@ public class AuditorPlacementPolicyCheckTest extends BookKeeperClusterTestCase {
for (int i = 0; i < numOfBookies; i++) {
BookieSocketAddress bookieAddress = new BookieSocketAddress("98.98.98." + i, 2181);
bookieAddresses.add(bookieAddress);
- regManager.registerBookie(bookieAddress.toString(), false);
+ regManager.registerBookie(bookieAddress.toString(), false, BookieServiceInfo.EMPTY);
}
LedgerManagerFactory mFactory = driver.getLedgerManagerFactory();
@@ -428,7 +429,7 @@ public class AuditorPlacementPolicyCheckTest extends BookKeeperClusterTestCase {
for (int i = 0; i < numOfBookies; i++) {
BookieSocketAddress bookieAddress = new BookieSocketAddress("98.98.98." + i, 2181);
bookieAddresses.add(bookieAddress);
- regManager.registerBookie(bookieAddress.toString(), false);
+ regManager.registerBookie(bookieAddress.toString(), false, BookieServiceInfo.EMPTY);
}
// only three racks
@@ -534,7 +535,7 @@ public class AuditorPlacementPolicyCheckTest extends BookKeeperClusterTestCase {
for (int i = 0; i < numOfBookies; i++) {
BookieSocketAddress bookieAddress = new BookieSocketAddress("98.98.98." + i, 2181);
bookieAddresses.add(bookieAddress);
- regManager.registerBookie(bookieAddress.toString(), false);
+ regManager.registerBookie(bookieAddress.toString(), false, BookieServiceInfo.EMPTY);
String zone = "/zone" + (i % 3);
String upgradeDomain = "/ud" + (i % 2);
String networkLocation = zone + upgradeDomain;
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorReplicasCheckTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorReplicasCheckTest.java
index a2d5f6b..85b2015 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorReplicasCheckTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorReplicasCheckTest.java
@@ -45,6 +45,7 @@ import org.apache.bookkeeper.client.LedgerMetadataBuilder;
import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.client.api.LedgerMetadata;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
@@ -185,7 +186,7 @@ public class AuditorReplicasCheckTest extends BookKeeperClusterTestCase {
for (int i = 0; i < numOfBookies; i++) {
bookieAddress = new BookieSocketAddress("98.98.98." + i, 2181);
bookieAddresses.add(bookieAddress);
- regManager.registerBookie(bookieAddress.toString(), false);
+ regManager.registerBookie(bookieAddress.toString(), false, BookieServiceInfo.EMPTY);
}
return bookieAddresses;
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/TestMain.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/TestMain.java
index d5253fd..24bce2e 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/TestMain.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/TestMain.java
@@ -26,12 +26,15 @@ import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import static org.powermock.api.mockito.PowerMockito.whenNew;
import java.io.IOException;
+import java.util.function.Supplier;
import org.apache.bookkeeper.common.component.LifecycleComponentStack;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.server.component.ServerLifecycleComponent;
import org.apache.bookkeeper.server.conf.BookieConfiguration;
@@ -80,9 +83,12 @@ public class TestMain {
BookieServer mockServer = PowerMockito.mock(BookieServer.class);
whenNew(BookieServer.class)
- .withArguments(any(ServerConfiguration.class), any(StatsLogger.class))
+ .withArguments(any(ServerConfiguration.class), any(StatsLogger.class), any(Supplier.class))
.thenReturn(mockServer);
+ BookieSocketAddress bookieAddress = new BookieSocketAddress("127.0.0.1", 1281);
+ when(mockServer.getLocalAddress()).thenReturn(bookieAddress);
+
LifecycleComponentStack stack = buildBookieServer(conf);
assertEquals(3, stack.getNumComponents());
assertTrue(stack.getComponent(2) instanceof TestComponent);
@@ -107,9 +113,12 @@ public class TestMain {
BookieServer mockServer = PowerMockito.mock(BookieServer.class);
whenNew(BookieServer.class)
- .withArguments(any(ServerConfiguration.class), any(StatsLogger.class))
+ .withArguments(any(ServerConfiguration.class), any(StatsLogger.class), any(Supplier.class))
.thenReturn(mockServer);
+ BookieSocketAddress bookieAddress = new BookieSocketAddress("127.0.0.1", 1281);
+ when(mockServer.getLocalAddress()).thenReturn(bookieAddress);
+
LifecycleComponentStack stack = buildBookieServer(conf);
assertEquals(2, stack.getNumComponents());
@@ -133,9 +142,12 @@ public class TestMain {
BookieServer mockServer = PowerMockito.mock(BookieServer.class);
whenNew(BookieServer.class)
- .withArguments(any(ServerConfiguration.class), any(StatsLogger.class))
+ .withArguments(any(ServerConfiguration.class), any(StatsLogger.class), any(Supplier.class))
.thenReturn(mockServer);
+ BookieSocketAddress bookieAddress = new BookieSocketAddress("127.0.0.1", 1281);
+ when(mockServer.getLocalAddress()).thenReturn(bookieAddress);
+
try {
buildBookieServer(conf);
fail("Should fail to start bookie server if `ignoreExtraServerComponentsStartupFailures` is set to false");
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 b8e2426..3d9e88f 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
@@ -41,6 +41,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieException;
@@ -50,6 +51,7 @@ import org.apache.bookkeeper.conf.AbstractConfiguration;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.metastore.InMemoryMetaStore;
import org.apache.bookkeeper.net.BookieSocketAddress;
@@ -648,7 +650,8 @@ public abstract class BookKeeperClusterTestCase {
protected BookieServer startBookie(ServerConfiguration conf)
throws Exception {
TestStatsProvider provider = new TestStatsProvider();
- BookieServer server = new BookieServer(conf, provider.getStatsLogger(""));
+ BookieServer server = new BookieServer(conf, provider.getStatsLogger(""),
+ BookieServiceInfo.NO_INFO);
BookieSocketAddress address = Bookie.getBookieAddress(conf);
bsLoggers.put(address, provider);
@@ -682,9 +685,11 @@ public abstract class BookKeeperClusterTestCase {
protected BookieServer startBookie(ServerConfiguration conf, final Bookie b)
throws Exception {
TestStatsProvider provider = new TestStatsProvider();
- BookieServer server = new BookieServer(conf, provider.getStatsLogger("")) {
+ BookieServer server = new BookieServer(conf, provider.getStatsLogger(""),
+ BookieServiceInfo.NO_INFO) {
@Override
- protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator) {
+ protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator,
+ Supplier<BookieServiceInfo> s) {
return b;
}
};
diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationManager.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationManager.java
index 87c831c..10015ad 100644
--- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationManager.java
+++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationManager.java
@@ -68,6 +68,7 @@ import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.bookie.BookieException.BookieIllegalOpException;
import org.apache.bookkeeper.bookie.BookieException.CookieNotFoundException;
import org.apache.bookkeeper.bookie.BookieException.MetadataStoreException;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerLayout;
import org.apache.bookkeeper.versioning.LongVersion;
@@ -145,7 +146,8 @@ class EtcdRegistrationManager implements RegistrationManager {
}
@Override
- public void registerBookie(String bookieId, boolean readOnly) throws BookieException {
+ public void registerBookie(String bookieId, boolean readOnly,
+ BookieServiceInfo bookieServiceInfo) throws BookieException {
if (readOnly) {
doRegisterReadonlyBookie(bookieId, bkRegister.get());
} else {
diff --git a/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationTest.java b/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationTest.java
index f36a496..0581a2c 100644
--- a/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationTest.java
+++ b/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationTest.java
@@ -42,6 +42,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.bookie.BookieException.MetadataStoreException;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.RegistrationClient;
import org.apache.bookkeeper.discover.RegistrationClient.RegistrationListener;
import org.apache.bookkeeper.metadata.etcd.testing.EtcdTestBase;
@@ -218,7 +219,7 @@ public class EtcdRegistrationTest extends EtcdTestBase {
Client client = newEtcdClient();
EtcdRegistrationManager regMgr = new EtcdRegistrationManager(client, scope, ttlSeconds);
bookies.add(regMgr);
- regMgr.registerBookie(newBookie(i), readonly);
+ regMgr.registerBookie(newBookie(i), readonly, BookieServiceInfo.EMPTY);
}
return bookies;
}
@@ -237,7 +238,7 @@ public class EtcdRegistrationTest extends EtcdTestBase {
try (EtcdRegistrationManager regManager = new EtcdRegistrationManager(
newEtcdClient(), scope, ttlSeconds)
) {
- regManager.registerBookie(bookieId, false);
+ regManager.registerBookie(bookieId, false, BookieServiceInfo.EMPTY);
leaseId = regManager.getBkRegister().getLeaseId();
log.info("Registered bookie under scope '{}' with lease = {}", scope, leaseId);
}
@@ -246,7 +247,7 @@ public class EtcdRegistrationTest extends EtcdTestBase {
try (EtcdRegistrationManager regManager = new EtcdRegistrationManager(
newEtcdClient(), scope, 100000 * ttlSeconds)
) {
- regManager.registerBookie(bookieId, false);
+ regManager.registerBookie(bookieId, false, BookieServiceInfo.EMPTY);
leaseId = regManager.getBkRegister().getLeaseId();
log.info("Registered bookie under scope '{}' with new lease = {}", scope, leaseId);
}
@@ -261,7 +262,7 @@ public class EtcdRegistrationTest extends EtcdTestBase {
try (EtcdRegistrationManager regManager = new EtcdRegistrationManager(
newEtcdClient(), scope, 10000000 * ttlSeconds)
) {
- regManager.registerBookie(bookieId, false);
+ regManager.registerBookie(bookieId, false, BookieServiceInfo.EMPTY);
leaseId = regManager.getBkRegister().getLeaseId();
log.info("Registered bookie under scope '{}' with lease = {}", scope, leaseId);
}
@@ -269,7 +270,7 @@ public class EtcdRegistrationTest extends EtcdTestBase {
try (EtcdRegistrationManager regManager = new EtcdRegistrationManager(
newEtcdClient(), scope, ttlSeconds)
) {
- regManager.registerBookie(bookieId, false);
+ regManager.registerBookie(bookieId, false, BookieServiceInfo.EMPTY);
fail("Should fail to register bookie under scope '{}'"
+ " since previous registration has not been expired yet");
} catch (MetadataStoreException mse) {
@@ -295,14 +296,14 @@ public class EtcdRegistrationTest extends EtcdTestBase {
try (EtcdRegistrationManager regManager = new EtcdRegistrationManager(
newEtcdClient(), scope, 10000000 * ttlSeconds)
) {
- regManager.registerBookie(bookieId, readonly);
+ regManager.registerBookie(bookieId, readonly, BookieServiceInfo.EMPTY);
leaseId = regManager.getBkRegister().getLeaseId();
log.info("Registered bookie under scope '{}' with lease = {}", scope, leaseId);
log.info("Trying to register using same lease '{}'", leaseId);
try (EtcdRegistrationManager regManager2 = new EtcdRegistrationManager(
regManager.getClient(), scope, regManager.getBkRegister()
)) {
- regManager.registerBookie(bookieId, readonly);
+ regManager.registerBookie(bookieId, readonly, BookieServiceInfo.EMPTY);
}
}
}
@@ -341,7 +342,7 @@ public class EtcdRegistrationTest extends EtcdTestBase {
log.info("before registration : bookies = {}", bookies);
assertEquals(0, bookies.size());
// registered
- regMgr.registerBookie(bookieId, readonly);
+ regMgr.registerBookie(bookieId, readonly, BookieServiceInfo.EMPTY);
bookies = getBookies(readonly);
log.info("after registered: bookies = {}", bookies);
assertEquals(1, bookies.size());
@@ -388,7 +389,7 @@ public class EtcdRegistrationTest extends EtcdTestBase {
)) {
try {
startBarrier.await();
- regMgr.registerBookie(bookieId, readonly);
+ regMgr.registerBookie(bookieId, readonly, BookieServiceInfo.EMPTY);
numSuccesses.incrementAndGet();
} catch (InterruptedException e) {
log.warn("Interrupted at waiting for the other threads to start", e);
diff --git a/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/testing/EtcdBKClusterTestBase.java b/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/testing/EtcdBKClusterTestBase.java
index ccc9a1e..7ccb0f0 100644
--- a/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/testing/EtcdBKClusterTestBase.java
+++ b/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/testing/EtcdBKClusterTestBase.java
@@ -27,6 +27,7 @@ import org.apache.bookkeeper.common.net.ServiceURI;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.meta.MetadataDrivers;
import org.apache.bookkeeper.metadata.etcd.EtcdMetadataBookieDriver;
import org.apache.bookkeeper.metadata.etcd.EtcdMetadataClientDriver;
@@ -120,7 +121,8 @@ public abstract class EtcdBKClusterTestBase extends EtcdTestBase {
private static BookieServer startBookie(ServerConfiguration conf) throws Exception {
conf.setAutoRecoveryDaemonEnabled(true);
TestStatsProvider provider = new TestStatsProvider();
- BookieServer server = new BookieServer(conf, provider.getStatsLogger(""));
+ BookieServer server = new BookieServer(conf, provider.getStatsLogger(""),
+ BookieServiceInfo.NO_INFO);
server.start();
return server;
}
diff --git a/site/_data/cli/shell.yaml b/site/_data/cli/shell.yaml
index 092c8b1..071fb34 100644
--- a/site/_data/cli/shell.yaml
+++ b/site/_data/cli/shell.yaml
@@ -40,6 +40,8 @@ commands:
description: Ledger ID
- flag: -force
description: Whether to force delete the Ledger without prompt..?
+- name: endpointinfo
+ description: Get endpoints of a Bookie.
- name: expandstorage
description: Add new empty ledger/index directories. Update the directories info in the conf file before running the command.
- name: help
diff --git a/stream/bk-grpc-name-resolver/src/test/java/org/apache/bookkeeper/grpc/resolver/BKRegistrationNameResolverTest.java b/stream/bk-grpc-name-resolver/src/test/java/org/apache/bookkeeper/grpc/resolver/BKRegistrationNameResolverTest.java
index 726c8b0..f877aa7 100644
--- a/stream/bk-grpc-name-resolver/src/test/java/org/apache/bookkeeper/grpc/resolver/BKRegistrationNameResolverTest.java
+++ b/stream/bk-grpc-name-resolver/src/test/java/org/apache/bookkeeper/grpc/resolver/BKRegistrationNameResolverTest.java
@@ -37,6 +37,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;
import lombok.Cleanup;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.meta.MetadataBookieDriver;
import org.apache.bookkeeper.meta.MetadataDrivers;
import org.apache.bookkeeper.stats.NullStatsLogger;
@@ -102,7 +103,7 @@ public class BKRegistrationNameResolverTest extends BookKeeperClusterTestCase {
InetSocketAddress address = new InetSocketAddress("127.0.0.1", 3181 + i);
addressSet.add(address);
bookieDriver.getRegistrationManager().registerBookie(
- "127.0.0.1:" + (3181 + i), false
+ "127.0.0.1:" + (3181 + i), false, BookieServiceInfo.EMPTY
);
}
@@ -137,7 +138,7 @@ public class BKRegistrationNameResolverTest extends BookKeeperClusterTestCase {
InetSocketAddress address = new InetSocketAddress("127.0.0.1", 3181 + i);
addressSet.add(address);
bookieDriver.getRegistrationManager().registerBookie(
- "127.0.0.1:" + (3181 + i), false
+ "127.0.0.1:" + (3181 + i), false, BookieServiceInfo.EMPTY
);
}
diff --git a/stream/distributedlog/core/src/test/java/org/apache/distributedlog/TestTxnId.java b/stream/distributedlog/core/src/test/java/org/apache/distributedlog/TestTxnId.java
index d5f4115..5e0a514 100644
--- a/stream/distributedlog/core/src/test/java/org/apache/distributedlog/TestTxnId.java
+++ b/stream/distributedlog/core/src/test/java/org/apache/distributedlog/TestTxnId.java
@@ -21,6 +21,7 @@ import java.io.File;
import java.util.ArrayList;
import java.util.List;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.stats.NullStatsProvider;
import org.junit.Assert;
@@ -107,7 +108,8 @@ public class TestTxnId extends TestDistributedLogBase {
conf.setJournalDirName(journalDir.getPath());
conf.setLedgerDirNames(new String[] { ledgerDir.getPath() });
- BookieServer server = new BookieServer(conf, new NullStatsProvider().getStatsLogger(""));
+ BookieServer server = new BookieServer(conf, new NullStatsProvider().getStatsLogger(""),
+ BookieServiceInfo.NO_INFO);
server.start();
while (!server.isRunning()) {
diff --git a/stream/server/src/main/java/org/apache/bookkeeper/stream/server/StorageServer.java b/stream/server/src/main/java/org/apache/bookkeeper/stream/server/StorageServer.java
index 4385d56..fe520e6 100644
--- a/stream/server/src/main/java/org/apache/bookkeeper/stream/server/StorageServer.java
+++ b/stream/server/src/main/java/org/apache/bookkeeper/stream/server/StorageServer.java
@@ -22,16 +22,21 @@ import java.io.File;
import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.UnknownHostException;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.clients.config.StorageClientSettings;
import org.apache.bookkeeper.clients.impl.channel.StorageServerChannel;
import org.apache.bookkeeper.clients.impl.internal.StorageServerClientManagerImpl;
+import org.apache.bookkeeper.common.component.ComponentInfoPublisher;
import org.apache.bookkeeper.common.component.ComponentStarter;
import org.apache.bookkeeper.common.component.LifecycleComponent;
import org.apache.bookkeeper.common.component.LifecycleComponentStack;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.statelib.impl.rocksdb.checkpoint.dlog.DLCheckpointStore;
import org.apache.bookkeeper.stats.NullStatsLogger;
@@ -182,9 +187,14 @@ public class StorageServer {
boolean startBookieAndStartProvider,
StatsLogger externalStatsLogger)
throws ConfigurationException, UnknownHostException {
+ final ComponentInfoPublisher componentInfoPublisher = new ComponentInfoPublisher();
+
+ final Supplier<BookieServiceInfo> bookieServiceInfoProvider =
+ () -> buildBookieServiceInfo(componentInfoPublisher);
LifecycleComponentStack.Builder serverBuilder = LifecycleComponentStack.newBuilder()
- .withName("storage-server");
+ .withName("storage-server")
+ .withComponentInfoPublisher(componentInfoPublisher);
BookieConfiguration bkConf = BookieConfiguration.of(conf);
bkConf.validate();
@@ -224,7 +234,7 @@ public class StorageServer {
// Create the bookie service
ServerConfiguration bkServerConf;
if (startBookieAndStartProvider) {
- BookieService bookieService = new BookieService(bkConf, rootStatsLogger);
+ BookieService bookieService = new BookieService(bkConf, rootStatsLogger, bookieServiceInfoProvider);
serverBuilder.addComponent(bookieService);
bkServerConf = bookieService.serverConf();
} else {
@@ -358,4 +368,25 @@ public class StorageServer {
.build();
}
+ /**
+ * Create the {@link BookieServiceInfo} starting from the published endpoints.
+ *
+ * @see ComponentInfoPublisher
+ * @param componentInfoPublisher the endpoint publisher
+ * @return the created bookie service info
+ */
+ private static BookieServiceInfo buildBookieServiceInfo(ComponentInfoPublisher componentInfoPublisher) {
+ List<BookieServiceInfo.Endpoint> endpoints = componentInfoPublisher.getEndpoints().values()
+ .stream().map(e -> {
+ return new BookieServiceInfo.Endpoint(
+ e.getId(),
+ e.getPort(),
+ e.getHost(),
+ e.getProtocol(),
+ e.getAuth(),
+ e.getExtensions()
+ );
+ }).collect(Collectors.toList());
+ return new BookieServiceInfo(componentInfoPublisher.getProperties(), endpoints);
+ }
}
diff --git a/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/BookieService.java b/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/BookieService.java
index 6f83752..2a41421 100644
--- a/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/BookieService.java
+++ b/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/BookieService.java
@@ -18,11 +18,13 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.function.Supplier;
import lombok.Getter;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.common.component.AbstractLifecycleComponent;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.stream.server.conf.BookieConfiguration;
@@ -38,11 +40,15 @@ public class BookieService extends AbstractLifecycleComponent<BookieConfiguratio
@Getter
private final ServerConfiguration serverConf;
private BookieServer bs;
+ private final Supplier<BookieServiceInfo> bookieServiceInfoProvider;
- public BookieService(BookieConfiguration conf, StatsLogger statsLogger) {
+
+ public BookieService(BookieConfiguration conf, StatsLogger statsLogger,
+ Supplier<BookieServiceInfo> bookieServiceInfoProvider) {
super("bookie-server", conf, statsLogger);
this.serverConf = new ServerConfiguration();
this.serverConf.loadConf(conf.getUnderlyingConf());
+ this.bookieServiceInfoProvider = bookieServiceInfoProvider;
}
@Override
@@ -61,7 +67,7 @@ public class BookieService extends AbstractLifecycleComponent<BookieConfiguratio
Arrays.asList(serverConf.getLedgerDirs()),
indexDirs);
try {
- this.bs = new BookieServer(serverConf, statsLogger);
+ this.bs = new BookieServer(serverConf, statsLogger, bookieServiceInfoProvider);
bs.start();
log.info("Started bookie server successfully.");
} catch (Exception e) {
diff --git a/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/RegistrationStateService.java b/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/RegistrationStateService.java
index 40261c0..2485c17 100644
--- a/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/RegistrationStateService.java
+++ b/stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/RegistrationStateService.java
@@ -25,6 +25,7 @@ import org.apache.bookkeeper.bookie.BookieStateManager;
import org.apache.bookkeeper.clients.utils.NetUtils;
import org.apache.bookkeeper.common.component.AbstractLifecycleComponent;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.discover.ZKRegistrationManager;
import org.apache.bookkeeper.stats.StatsLogger;
@@ -79,7 +80,8 @@ public class RegistrationStateService
statsLogger.scope("state"),
() -> regManager,
Collections.emptyList(),
- () -> NetUtils.endpointToString(myEndpoint));
+ () -> NetUtils.endpointToString(myEndpoint),
+ BookieServiceInfo.NO_INFO);
stateManager.initState();
stateManager.registerBookie(true).get();
log.info("Successfully register myself under registration path {}/{}",
diff --git a/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/commands/BookieCommandGroup.java b/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/commands/BookieCommandGroup.java
index 02ccf47..54e9f7c 100644
--- a/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/commands/BookieCommandGroup.java
+++ b/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/commands/BookieCommandGroup.java
@@ -38,6 +38,7 @@ import org.apache.bookkeeper.tools.cli.commands.bookie.ReadLogMetadataCommand;
import org.apache.bookkeeper.tools.cli.commands.bookie.RebuildDBLedgerLocationsIndexCommand;
import org.apache.bookkeeper.tools.cli.commands.bookie.RegenerateInterleavedStorageIndexFileCommand;
import org.apache.bookkeeper.tools.cli.commands.bookie.SanityTestCommand;
+import org.apache.bookkeeper.tools.cli.commands.bookies.EndpointInfoCommand;
import org.apache.bookkeeper.tools.common.BKFlags;
import org.apache.bookkeeper.tools.framework.CliCommandGroup;
import org.apache.bookkeeper.tools.framework.CliSpec;
@@ -58,6 +59,7 @@ public class BookieCommandGroup extends CliCommandGroup<BKFlags> {
.addCommand(new LastMarkCommand())
.addCommand(new InitCommand())
.addCommand(new FormatCommand())
+ .addCommand(new EndpointInfoCommand())
.addCommand(new SanityTestCommand())
.addCommand(new LedgerCommand())
.addCommand(new ListFilesOnDiscCommand())