diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerClientIntegrationTest.java index bcde1603f49fd..0c105b5dcc965 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerClientIntegrationTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerClientIntegrationTest.java @@ -21,10 +21,14 @@ import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotEquals; import static org.testng.Assert.assertNull; import static org.testng.Assert.assertTrue; import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.net.URI; import java.util.ArrayList; import java.util.List; import java.util.NavigableMap; @@ -467,4 +471,55 @@ public long getTimestamp() { } } + /** + * 
+     * It verifies that client closes lookup connection after configured lifetime.
+     * 1. creates client with different lookup-url(pulsar://0.0.0.0:$port) than brokerServiceUrl(pulsar://localhost:$port)
+     * 2. verifies lookup-connection gets closed after configured lifetime
+     * 3. verifies created consumer's connection should not be affected and should be live
+     *
+ * + * @throws Exception + */ + @Test + public void testLookupConnectionLifeTime() throws Exception { + final String topicName = "persistent://prop/usw/my-ns/newTopic"; + final long connectionLifeTimeInSecods = 1; + com.yahoo.pulsar.client.api.ClientConfiguration clientConf = new com.yahoo.pulsar.client.api.ClientConfiguration(); + clientConf.setStatsInterval(0, TimeUnit.SECONDS); + clientConf.setLookupConnectionLifetimeInSecond(connectionLifeTimeInSecods); + // to specify two separate connection url => producer/consumer will be crreated on connection with + // brokerServiceUrl= "localhost:" and lookup-connection will be created using "0.0.0.0:" + final String lookupUrlStr = "pulsar://0.0.0.0:"; + URI lookupUrl = new URI(lookupUrlStr + BROKER_PORT); + PulsarClientImpl pulsarClient = (PulsarClientImpl) PulsarClient.create(lookupUrl.toString(), clientConf); + boolean isLookupAndServiceUrlSame = false; + ConsumerImpl consumer = null; + try { + consumer = (ConsumerImpl) pulsarClient.subscribe(topicName, "mysub", new ConsumerConfiguration()); + } catch (PulsarClientException connectionException) { + log.error("failed to lookup on {}. Now, connecting on {}", lookupUrlStr, pulsar.getBrokerServiceUrl()); + // if routing failed on : 0.0.0.0 then create connection on the same brokerServiceUrl + pulsarClient = (PulsarClientImpl) PulsarClient.create(pulsar.getBrokerServiceUrl(), clientConf); + consumer = (ConsumerImpl) pulsarClient.subscribe(topicName, "mysub", new ConsumerConfiguration()); + isLookupAndServiceUrlSame = true; + } + InetSocketAddress lookupConnectionAddress = new InetSocketAddress(lookupUrl.getHost(), lookupUrl.getPort()); + ClientCnx lookupConnection = pulsarClient.getCnxPool().getConnection(lookupConnectionAddress).get(); + ClientCnx consumerConnection = consumer.getClientCnx(); + assertTrue(lookupConnection.channel().isActive()); + if (!isLookupAndServiceUrlSame) { + assertNotEquals(lookupConnection, consumerConnection); + assertTrue(consumerConnection.channel().isActive()); + } + + // wait connection to be closed + Thread.sleep((connectionLifeTimeInSecods * 1000) + 1000); + // verify lookup-connection should be closed + assertFalse(lookupConnection.channel().isActive()); + if (!isLookupAndServiceUrlSame) { + assertTrue(consumerConnection.channel().isActive()); + } + } + } diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/ClientConfiguration.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/ClientConfiguration.java index 6a2ab81035b4a..fba6fb259f9f2 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/ClientConfiguration.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/ClientConfiguration.java @@ -42,6 +42,7 @@ public class ClientConfiguration implements Serializable { private int numIoThreads = 1; private int numListenerThreads = 1; private int connectionsPerBroker = 1; + private long lookupConnectionLifetimeInSecond = 600; private boolean useTcpNoDelay = true; @@ -219,6 +220,30 @@ public void setConnectionsPerBroker(int connectionsPerBroker) { this.connectionsPerBroker = connectionsPerBroker; } + /** + * + * @return lifetime of the persistent-connection used for the binary-proto lookup + */ + public long getLookupConnectionLifetimeInSecond() { + return lookupConnectionLifetimeInSecond; + } + + /** + * Sets lifetime of the persistent-connection (with broker) used for the binary-proto lookup (default: 10 + * mins) + *

+ * Setting time to -1 will not force connection to be closed after default-lifetime duration. Client creates a + * separate dedicated connection with broker for a given service-lookup url, if this url is same as broker's + * brokerServiceUrl then client shares the same created lookup-connection for producer/consumer connection also. So, + * in that case {@link #setLookupConnectionLifetimeInSecond(long)} should be configured -1 to avoid + * producer/consumer disruption due to disconnection/reconnection. + * + * @param lookupConnectionLifetimeInSecond + */ + public void setLookupConnectionLifetimeInSecond(long lookupConnectionLifetimeInSecond) { + this.lookupConnectionLifetimeInSecond = lookupConnectionLifetimeInSecond; + } + /** * @return whether TCP no-delay should be set on the connections */ @@ -322,8 +347,8 @@ public int getConcurrentLookupRequest() { /** * Number of concurrent lookup-requests allowed on each broker-connection to prevent overload on broker. - * (default: 5000) It should be configured with higher value only in case of it requires to produce/subscribe on - * thousands of topic using created {@link PulsarClient} + * (default: 5000) It should be configured with higher value only in case of it requires to produce/subscribe + * on thousands of topic using created {@link PulsarClient} * * @param concurrentLookupRequest */ diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BinaryProtoLookupService.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BinaryProtoLookupService.java index 98e905ad2532c..5f16464eeb648 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BinaryProtoLookupService.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BinaryProtoLookupService.java @@ -36,6 +36,7 @@ class BinaryProtoLookupService implements LookupService { private final PulsarClientImpl client; protected final InetSocketAddress serviceAddress; private final boolean useTls; + private final long lookupConnectionLifetimeInSecond; public BinaryProtoLookupService(PulsarClientImpl client, String serviceUrl, boolean useTls) throws PulsarClientException { @@ -45,6 +46,7 @@ public BinaryProtoLookupService(PulsarClientImpl client, String serviceUrl, bool try { uri = new URI(serviceUrl); this.serviceAddress = new InetSocketAddress(uri.getHost(), uri.getPort()); + this.lookupConnectionLifetimeInSecond = client.getConfiguration().getLookupConnectionLifetimeInSecond(); } catch (Exception e) { log.error("Invalid service-url {} provided {}", serviceUrl, e.getMessage(), e); throw new PulsarClientException.InvalidServiceURL(e); @@ -74,7 +76,7 @@ private CompletableFuture findBroker(InetSocketAddress socket DestinationName destination) { CompletableFuture addressFuture = new CompletableFuture(); - client.getCnxPool().getConnection(socketAddress).thenAccept(clientCnx -> { + client.getCnxPool().getConnection(socketAddress, lookupConnectionLifetimeInSecond).thenAccept(clientCnx -> { long requestId = client.newRequestId(); ByteBuf request = Commands.newLookup(destination.toString(), authoritative, requestId); clientCnx.newLookup(request, requestId).thenAccept(lookupDataResult -> { @@ -132,7 +134,7 @@ private CompletableFuture getPartitionedTopicMetadata( CompletableFuture partitionFuture = new CompletableFuture(); - client.getCnxPool().getConnection(socketAddress).thenAccept(clientCnx -> { + client.getCnxPool().getConnection(socketAddress, lookupConnectionLifetimeInSecond).thenAccept(clientCnx -> { long requestId = client.newRequestId(); ByteBuf request = Commands.newPartitionMetadataRequest(destination.toString(), requestId); clientCnx.newLookup(request, requestId).thenAccept(lookupDataResult -> { diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java index 0953161c9b044..4f755a1209cb7 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java @@ -61,6 +61,8 @@ public class ClientCnx extends PulsarHandler { private final CompletableFuture connectionFuture = new CompletableFuture(); private final Semaphore pendingLookupRequestSemaphore; + // Timestamp of when this connection used for lookup + protected volatile long lastLookupRequestTime; enum State { None, SentConnectFrame, Ready @@ -258,6 +260,7 @@ protected void handlePartitionResponse(CommandPartitionedTopicMetadataResponse l private boolean addPendingLookupRequests(long requestId, CompletableFuture future) { if (pendingLookupRequestSemaphore.tryAcquire()) { pendingLookupRequests.put(requestId, future); + lastLookupRequestTime = System.nanoTime(); return true; } return false; @@ -338,6 +341,7 @@ CompletableFuture newLookup(ByteBuf request, long requestId) { if (!writeFuture.isSuccess()) { log.warn("{} Failed to send request {} to broker: {}", ctx.channel(), requestId, writeFuture.cause().getMessage()); + getAndRemovePendingLookupRequest(requestId); future.completeExceptionally(writeFuture.cause()); } }); @@ -377,6 +381,7 @@ CompletableFuture sendRequestWithId(ByteBuf cmd, long requestId) { ctx.writeAndFlush(cmd).addListener(writeFuture -> { if (!writeFuture.isSuccess()) { log.warn("{} Failed to send request to broker: {}", ctx.channel(), writeFuture.cause().getMessage()); + pendingRequests.remove(requestId); future.completeExceptionally(writeFuture.cause()); } }); diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConnectionPool.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConnectionPool.java index 5741a0e27225e..7b70563924347 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConnectionPool.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConnectionPool.java @@ -24,6 +24,9 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.SystemUtils; import org.slf4j.Logger; @@ -55,6 +58,7 @@ public class ConnectionPool implements Closeable { private final Bootstrap bootstrap; private final EventLoopGroup eventLoopGroup; private final int maxConnectionsPerHosts; + private final ScheduledExecutorService scheduler; private static final int MaxMessageSize = 5 * 1024 * 1024; public static final String TLS_HANDLER = "tls"; @@ -62,6 +66,7 @@ public class ConnectionPool implements Closeable { public ConnectionPool(final PulsarClientImpl client, EventLoopGroup eventLoopGroup) { this.eventLoopGroup = eventLoopGroup; this.maxConnectionsPerHosts = client.getConfiguration().getConnectionsPerBroker(); + this.scheduler = Executors.newScheduledThreadPool(1); pool = new ConcurrentHashMap<>(); bootstrap = new Bootstrap(); @@ -111,22 +116,36 @@ public void initChannel(SocketChannel ch) throws Exception { private static final Random random = new Random(); public CompletableFuture getConnection(final InetSocketAddress address) { + return getConnection(address, -1); + } + + /** + * + * @param address + * remote client address + * @param connectionLifetimeInSecond + * connection lifetime (in second): if it's > 0 then created connection will be automatically closed + * after given connectionLifetimeInSecond + * @return + */ + public CompletableFuture getConnection(final InetSocketAddress address, + long connectionLifetimeInSecond) { if (maxConnectionsPerHosts == 0) { // Disable pooling - return createConnection(address, -1); + return createConnection(address, -1, connectionLifetimeInSecond); } final int randomKey = signSafeMod(random.nextInt(), maxConnectionsPerHosts); return pool.computeIfAbsent(address, a -> new ConcurrentHashMap<>()) // - .computeIfAbsent(randomKey, k -> createConnection(address, randomKey)); + .computeIfAbsent(randomKey, k -> createConnection(address, randomKey, connectionLifetimeInSecond)); } - private CompletableFuture createConnection(InetSocketAddress address, int connectionKey) { + private CompletableFuture createConnection(InetSocketAddress address, int connectionKey, + long connectionLifetimeInSecond) { if (log.isDebugEnabled()) { log.debug("Connection for {} not found in cache", address); } - final CompletableFuture cnxFuture = new CompletableFuture(); // Trigger async connect to broker @@ -170,14 +189,35 @@ private CompletableFuture createConnection(InetSocketAddress address, cnx.ctx().close(); return null; }); + + // close the connection after connection-lifetime completes + if (connectionLifetimeInSecond > 0) { + scheduleCloseConnection(cnx, connectionLifetimeInSecond); + } }); return cnxFuture; } + private void scheduleCloseConnection(ClientCnx cnx, long connectionLifetimeInSecond) { + scheduler.schedule(() -> { + if (cnx != null && cnx.channel().isActive()) { + if (System.nanoTime() - cnx.lastLookupRequestTime > TimeUnit.SECONDS + .toNanos(connectionLifetimeInSecond)) { + cnx.channel().disconnect(); + } else { + // connection is active as received lookup request in last connectionLifetimeInSecond, schedule + // retry and check later again + scheduleCloseConnection(cnx, connectionLifetimeInSecond); + } + } + }, connectionLifetimeInSecond, TimeUnit.SECONDS); + } + @Override public void close() throws IOException { eventLoopGroup.shutdownGracefully(); + scheduler.shutdown(); } private void cleanupConnection(InetSocketAddress address, int connectionKey,