[fix][client] Remove producer when close producer command is received (#16028)

Author: nodece

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

@@ -111,7 +111,8 @@ public class ClientCnx extends PulsarHandler {
     // LookupRequests that waiting in client side.
     private final Queue<Pair<Long, Pair<ByteBuf, TimedCompletableFuture<LookupDataResult>>>> waitingLookupRequests;
 
-    private final ConcurrentLongHashMap<ProducerImpl<?>> producers =
+    @VisibleForTesting
+    final ConcurrentLongHashMap<ProducerImpl<?>> producers =
             ConcurrentLongHashMap.<ProducerImpl<?>>newBuilder()
                     .expectedItems(16)
                     .concurrencyLevel(1)
@@ -721,7 +722,7 @@ protected void handleError(CommandError error) {
     protected void handleCloseProducer(CommandCloseProducer closeProducer) {
         log.info("[{}] Broker notification of Closed producer: {}", remoteAddress, closeProducer.getProducerId());
         final long producerId = closeProducer.getProducerId();
-        ProducerImpl<?> producer = producers.get(producerId);
+        ProducerImpl<?> producer = producers.remove(producerId);
         if (producer != null) {
             producer.connectionClosed(this);
         } else {

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

@@ -38,6 +38,7 @@
 import org.apache.pulsar.client.api.PulsarClientException.BrokerMetadataException;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.common.api.proto.CommandCloseConsumer;
+import org.apache.pulsar.common.api.proto.CommandCloseProducer;
 import org.apache.pulsar.common.api.proto.CommandError;
 import org.apache.pulsar.common.api.proto.ServerError;
 import org.apache.pulsar.common.protocol.Commands;
@@ -156,7 +157,7 @@ public void testGetLastMessageIdWithError() throws Exception {
 
     @Test
     public void testHandleCloseConsumer() {
-        ThreadFactory threadFactory = new DefaultThreadFactory("testReceiveErrorAtSendConnectFrameState");
+        ThreadFactory threadFactory = new DefaultThreadFactory("testHandleCloseConsumer");
         EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, threadFactory);
         ClientConfigurationData conf = new ClientConfigurationData();
         ClientCnx cnx = new ClientCnx(conf, eventLoop);
@@ -165,11 +166,28 @@ public void testHandleCloseConsumer() {
         cnx.registerConsumer(consumerId, mock(ConsumerImpl.class));
         assertEquals(cnx.consumers.size(), 1);
 
-        CommandCloseConsumer closeConsumer = new CommandCloseConsumer()
-                .setConsumerId(1);
+        CommandCloseConsumer closeConsumer = new CommandCloseConsumer().setConsumerId(consumerId);
         cnx.handleCloseConsumer(closeConsumer);
         assertEquals(cnx.consumers.size(), 0);
 
         eventLoop.shutdownGracefully();
     }
+
+    @Test
+    public void testHandleCloseProducer() {
+        ThreadFactory threadFactory = new DefaultThreadFactory("testHandleCloseProducer");
+        EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, threadFactory);
+        ClientConfigurationData conf = new ClientConfigurationData();
+        ClientCnx cnx = new ClientCnx(conf, eventLoop);
+
+        long producerId = 1;
+        cnx.registerProducer(producerId, mock(ProducerImpl.class));
+        assertEquals(cnx.producers.size(), 1);
+
+        CommandCloseProducer closeProducerCmd = new CommandCloseProducer().setProducerId(producerId);
+        cnx.handleCloseProducer(closeProducerCmd);
+        assertEquals(cnx.producers.size(), 0);
+
+        eventLoop.shutdownGracefully();
+    }
 }