The service uri should be pointed to the proxy service if {@link ConnectionMode#SNI_ROUTING} is configured.
+ *
+ * @return metadata service uri.
+ */
+ public String getBookieServiceUri() {
+ return getString(BK_SERVICE_URI);
+ }
+
+ /**
+ * Set the service uri to connect to bookies.
+ *
+ * @param serviceUri the bookie service uri.
+ * @return the configuration object.
+ * @see #getBookieServiceUri()
+ */
+ public T setBookieServiceUri(String serviceUri) {
+ setProperty(BK_SERVICE_URI, serviceUri);
+ return getThis();
+ }
+
+ /**
+ * Get the {@link ConnectionMode} that the bookkeeper client uses to connect to bookies.
+ *
+ * @return the {@link ConnectionMode} that the bookkeeper client used to connect to bookies.
+ */
+ public ConnectionMode getBookieConnectionMode() {
+ String modeStr = getString(BK_CONNECTION_MODE, ConnectionMode.DIRECT.name());
+ return ConnectionMode.valueOf(modeStr);
+ }
+
+ /**
+ * Set the {@link ConnectionMode} that the bookkeeper client uses to connect to bookies.
+ *
+ * @param connectionMode connection mode that the bookkeeper client uses to connect to bookies.
+ * @return the configuration object
+ * @see #getBookieConnectionMode()
+ * @see #getBookieServiceUri()
+ */
+ public T setBookieConnectionMode(ConnectionMode connectionMode) {
+ setProperty(BK_CONNECTION_MODE, connectionMode.name());
+ return getThis();
+ }
+
/**
* Get metadata service uri.
*
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
index 88239dcdfd8..f8a771cb6f1 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
@@ -92,7 +92,9 @@
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeperClientStats;
import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo;
+import org.apache.bookkeeper.client.api.ConnectionMode;
import org.apache.bookkeeper.client.api.WriteFlag;
+import org.apache.bookkeeper.common.net.ServiceURI;
import org.apache.bookkeeper.common.util.MdcUtils;
import org.apache.bookkeeper.common.util.OrderedExecutor;
import org.apache.bookkeeper.conf.ClientConfiguration;
@@ -346,6 +348,9 @@ enum ConnectionState {
private final SecurityHandlerFactory shFactory;
private volatile boolean isWritable = true;
private long lastBookieUnavailableLogTimestamp = 0;
+ private final ConnectionMode connectionMode;
+ private final BookieSocketAddress proxyAddress;
+ private volatile BookieSocketAddress targetBookieAddress;
public PerChannelBookieClient(OrderedExecutor executor, EventLoopGroup eventLoopGroup,
BookieId addr, BookieAddressResolver bookieAddressResolver) throws SecurityException {
@@ -377,7 +382,8 @@ public PerChannelBookieClient(ClientConfiguration conf, OrderedExecutor executor
EventLoopGroup eventLoopGroup,
ByteBufAllocator allocator,
BookieId bookieId,
- StatsLogger parentStatsLogger, ClientAuthProvider.Factory authProviderFactory,
+ StatsLogger parentStatsLogger,
+ ClientAuthProvider.Factory authProviderFactory,
ExtensionRegistry extRegistry,
PerChannelBookieClientPool pcbcPool,
SecurityHandlerFactory shFactory,
@@ -387,6 +393,19 @@ public PerChannelBookieClient(ClientConfiguration conf, OrderedExecutor executor
this.bookieId = bookieId;
this.bookieAddressResolver = bookieAddressResolver;
this.executor = executor;
+ this.connectionMode = conf.getBookieConnectionMode();
+ if (ConnectionMode.SNI_ROUTING == connectionMode) {
+ ServiceURI serviceURI = ServiceURI.create(conf.getBookieServiceUri());
+ String hostName = serviceURI.getServiceHosts()[0];
+ try {
+ this.proxyAddress = new BookieSocketAddress(hostName);
+ } catch (UnknownHostException e) {
+ throw new RuntimeException("Invalid bookie service uri : " + serviceURI, e);
+ }
+ } else {
+ this.proxyAddress = null;
+ }
+
if (LocalBookiesRegistry.isLocalBookie(bookieId)) {
this.eventLoopGroup = new DefaultEventLoopGroup();
} else {
@@ -538,9 +557,8 @@ protected ChannelFuture connect() {
if (LOG.isDebugEnabled()) {
LOG.debug("Connecting to bookie: {}", bookieId);
}
- BookieSocketAddress addr;
try {
- addr = bookieAddressResolver.resolve(bookieId);
+ targetBookieAddress = bookieAddressResolver.resolve(bookieId);
} catch (BookieAddressResolver.BookieIdNotResolvedException err) {
LOG.error("Cannot connect to {} as endpopint resolution failed", bookieId, err);
return processBookieNotResolvedError(startTime, err);
@@ -599,9 +617,15 @@ protected void initChannel(Channel ch) throws Exception {
}
});
- SocketAddress bookieAddr = addr.getSocketAddress();
+ SocketAddress bookieAddr;
if (eventLoopGroup instanceof DefaultEventLoopGroup) {
bookieAddr = new LocalAddress(bookieId.toString());
+ } else {
+ if (ConnectionMode.SNI_ROUTING == connectionMode) {
+ bookieAddr = proxyAddress.getSocketAddress();
+ } else {
+ bookieAddr = targetBookieAddress.getSocketAddress();
+ }
}
ChannelFuture future = bootstrap.connect(bookieAddr);
@@ -1488,7 +1512,20 @@ public String toString() {
void initTLSHandshake() {
// create TLS handler
PerChannelBookieClient parentObj = PerChannelBookieClient.this;
- SslHandler handler = parentObj.shFactory.newTLSHandler();
+ SslHandler handler;
+ BookieSocketAddress bookieSocketAddress = targetBookieAddress;
+ if (ConnectionMode.SNI_ROUTING == connectionMode) {
+ if (bookieSocketAddress != null) {
+ handler = parentObj.shFactory.newTLSHandler(
+ bookieSocketAddress.getHostName(), bookieSocketAddress.getPort());
+ } else {
+ LOG.warn("No target bookie address for BookieId [{}] is resolved"
+ + " when connecting to proxy [{}] using SNI routing", bookieId, proxyAddress);
+ handler = parentObj.shFactory.newTLSHandler();
+ }
+ } else {
+ handler = parentObj.shFactory.newTLSHandler();
+ }
channel.pipeline().addFirst(parentObj.shFactory.getHandlerName(), handler);
handler.handshakeFuture().addListener(new GenericFutureListener