close binary-proto-lookup connection after connection-lifetime

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() {
         }
     }
 
+    /**
+     * <pre>
+     * 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
+     * </pre>
+     * 
+     * @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:<port>" and lookup-connection will be created using "0.0.0.0:<port>"
+        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());
+        }
+    }
+
 }

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 <i>(default: 10
+     * mins)</i>
+     * <p>
+     * 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.
-     * <i>(default: 5000)</i> It should be configured with higher value only in case of it requires to produce/subscribe on
-     * thousands of topic using created {@link PulsarClient}
+     * <i>(default: 5000)</i> 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
      */

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<InetSocketAddress> findBroker(InetSocketAddress socket
             DestinationName destination) {
         CompletableFuture<InetSocketAddress> addressFuture = new CompletableFuture<InetSocketAddress>();
 
-        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<PartitionedTopicMetadata> getPartitionedTopicMetadata(
 
         CompletableFuture<PartitionedTopicMetadata> partitionFuture = new CompletableFuture<PartitionedTopicMetadata>();
 
-        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 -> {

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java

@@ -61,6 +61,8 @@ public class ClientCnx extends PulsarHandler {
 
     private final CompletableFuture<Void> connectionFuture = new CompletableFuture<Void>();
     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<LookupDataResult> future) {
         if (pendingLookupRequestSemaphore.tryAcquire()) {
             pendingLookupRequests.put(requestId, future);
+            lastLookupRequestTime = System.nanoTime();
             return true;
         }
         return false;
@@ -338,6 +341,7 @@ CompletableFuture<LookupDataResult> 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<String> 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());
             }
         });

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,13 +58,15 @@ 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";
 
     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<ClientCnx> 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<ClientCnx> 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<ClientCnx> createConnection(InetSocketAddress address, int connectionKey) {
+    private CompletableFuture<ClientCnx> createConnection(InetSocketAddress address, int connectionKey,
+            long connectionLifetimeInSecond) {
         if (log.isDebugEnabled()) {
             log.debug("Connection for {} not found in cache", address);
         }
-
         final CompletableFuture<ClientCnx> cnxFuture = new CompletableFuture<ClientCnx>();
 
         // Trigger async connect to broker
@@ -170,14 +189,35 @@ private CompletableFuture<ClientCnx> 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,