[ISSUE 9783][pulsar-client] Allow pulsar client receive external timer (

#9802)

Fixed #9783

Allow pulsar client to receive external timer instance

Add new constructor to provide an external timer, and share timer in pulsar proxy

- [x] Make sure that the change passes the CI checks.

(cherry picked from commit af6eaba)

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

@@ -94,6 +94,7 @@ public class PulsarClientImpl implements PulsarClient {
     private LookupService lookup;
     private final ConnectionPool cnxPool;
     private final Timer timer;
+    private boolean needStopTimer;
     private final ExecutorProvider externalExecutorProvider;
     private final ExecutorProvider internalExecutorService;
 
@@ -130,11 +131,16 @@ public PulsarClientImpl(ClientConfigurationData conf) throws PulsarClientExcepti
     }
 
     public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) throws PulsarClientException {
-        this(conf, eventLoopGroup, new ConnectionPool(conf, eventLoopGroup));
+        this(conf, eventLoopGroup, new ConnectionPool(conf, eventLoopGroup), null);
     }
 
     public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool)
             throws PulsarClientException {
+        this(conf, eventLoopGroup, cnxPool, null);
+    }
+
+    public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool, Timer timer)
+            throws PulsarClientException {
         if (conf == null || isBlank(conf.getServiceUrl()) || eventLoopGroup == null) {
             throw new PulsarClientException.InvalidConfigurationException("Invalid client configuration");
         }
@@ -151,7 +157,12 @@ public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGr
         } else {
             lookup = new BinaryProtoLookupService(this, conf.getServiceUrl(), conf.getListenerName(), conf.isUseTls(), externalExecutorProvider.getExecutor());
         }
-        timer = new HashedWheelTimer(getThreadFactory("pulsar-timer"), 1, TimeUnit.MILLISECONDS);
+        if (timer == null) {
+            this.timer = new HashedWheelTimer(getThreadFactory("pulsar-timer"), 1, TimeUnit.MILLISECONDS);
+            needStopTimer = true;
+        } else {
+            this.timer = timer;
+        }
         producers = Collections.newSetFromMap(new ConcurrentHashMap<>());
         consumers = Collections.newSetFromMap(new ConcurrentHashMap<>());
 
@@ -601,7 +612,9 @@ public void shutdown() throws PulsarClientException {
         try {
             lookup.close();
             cnxPool.close();
-            timer.stop();
+            if (needStopTimer) {
+                timer.stop();
+            }
             externalExecutorProvider.shutdownNow();
             internalExecutorService.shutdownNow();
             conf.getAuthentication().close();

pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java

@@ -22,6 +22,7 @@
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.verify;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotSame;
 import static org.testng.Assert.assertSame;
@@ -33,8 +34,10 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.EventLoopGroup;
+import io.netty.util.HashedWheelTimer;
 import io.netty.util.concurrent.DefaultThreadFactory;
 
+import java.lang.reflect.Field;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.ArrayList;
@@ -54,6 +57,7 @@
 import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
 import org.apache.pulsar.common.util.netty.EventLoopUtil;
 import org.mockito.internal.util.reflection.FieldSetter;
+import org.mockito.Mockito;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
@@ -146,4 +150,32 @@ public void testConsumerIsClosed() throws Exception {
                 assertSame(consumer.getState(), HandlerState.State.Closed));
     }
 
+    @Test
+    public void testInitializeWithoutTimer() throws Exception {
+        ClientConfigurationData conf = new ClientConfigurationData();
+        conf.setServiceUrl("pulsar://localhost:6650");
+        PulsarClientImpl client = new PulsarClientImpl(conf);
+
+        HashedWheelTimer timer = mock(HashedWheelTimer.class);
+        Field field = client.getClass().getDeclaredField("timer");
+        field.setAccessible(true);
+        field.set(client, timer);
+
+        client.shutdown();
+        verify(timer).stop();
+    }
+
+    @Test
+    public void testInitializeWithTimer() throws PulsarClientException {
+        ClientConfigurationData conf = new ClientConfigurationData();
+        EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, new DefaultThreadFactory("test"));
+        ConnectionPool pool = Mockito.spy(new ConnectionPool(conf, eventLoop));
+        conf.setServiceUrl("pulsar://localhost:6650");
+
+        HashedWheelTimer timer = new HashedWheelTimer();
+        PulsarClientImpl client = new PulsarClientImpl(conf, eventLoop, pool, timer);
+
+        client.shutdown();
+        client.timer().stop();
+    }
 }

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java

@@ -299,7 +299,7 @@ protected void handleConnect(CommandConnect connect) {
             if (!service.getConfiguration().isAuthenticationEnabled()) {
                 this.client = new PulsarClientImpl(clientConf, service.getWorkerGroup(),
                     new ProxyConnectionPool(clientConf, service.getWorkerGroup(),
-                        () -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersion)));
+                        () -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersion)), service.getTimer());
 
                 completeConnect();
                 return;
@@ -436,7 +436,7 @@ private PulsarClientImpl createClient(final ClientConfigurationData clientConf,
             final String clientAuthMethod, final int protocolVersion) throws PulsarClientException {
         return new PulsarClientImpl(clientConf, service.getWorkerGroup(),
                 new ProxyConnectionPool(clientConf, service.getWorkerGroup(), () -> new ProxyClientCnx(clientConf,
-                        service.getWorkerGroup(), clientAuthRole, clientAuthData, clientAuthMethod, protocolVersion)));
+                        service.getWorkerGroup(), clientAuthRole, clientAuthData, clientAuthMethod, protocolVersion)), service.getTimer());
     }
 
     private static int getProtocolVersionToAdvertise(CommandConnect connect) {

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java

@@ -27,6 +27,8 @@
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
 import io.netty.util.concurrent.DefaultThreadFactory;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timer;
 import io.prometheus.client.Counter;
 import io.prometheus.client.Gauge;
 import lombok.Getter;
@@ -67,6 +69,7 @@
 public class ProxyService implements Closeable {
 
     private final ProxyConfiguration proxyConfig;
+    private final Timer timer;
     private String serviceUrl;
     private String serviceUrlTls;
     private ConfigurationCacheService configurationCacheService;
@@ -124,6 +127,7 @@ public ProxyService(ProxyConfiguration proxyConfig,
                         AuthenticationService authenticationService) throws IOException {
         checkNotNull(proxyConfig);
         this.proxyConfig = proxyConfig;
+        this.timer = new HashedWheelTimer(new DefaultThreadFactory("pulsar-timer", Thread.currentThread().isDaemon()), 1, TimeUnit.MILLISECONDS);
         this.clientCnxs = Sets.newConcurrentHashSet();
         this.topicStats = Maps.newConcurrentMap();
 
@@ -243,6 +247,9 @@ public void close() throws IOException {
 
         acceptorGroup.shutdownGracefully();
         workerGroup.shutdownGracefully();
+        if (timer != null) {
+            timer.stop();
+        }
     }
 
     public String getServiceUrl() {
@@ -257,6 +264,10 @@ public ProxyConfiguration getConfiguration() {
         return proxyConfig;
     }
 
+    public Timer getTimer() {
+        return timer;
+    }
+
     public AuthenticationService getAuthenticationService() {
         return authenticationService;
     }