[fix][broker] Remove consumer from cache if closed before creation

Signed-off-by: Zixuan Liu <nodeces@gmail.com>

Author: nodece

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -2229,13 +2229,15 @@ protected void handleCloseConsumer(CommandCloseConsumer closeConsumer) {
             // create operation will complete, the new consumer will be discarded.
             log.info("[{}] Closed consumer before its creation was completed. consumerId={}",
                      remoteAddress, consumerId);
+            consumers.remove(consumerId, consumerFuture);
             commandSender.sendSuccessResponse(requestId);
             return;
         }
 
         if (consumerFuture.isCompletedExceptionally()) {
             log.info("[{}] Closed consumer that already failed to be created. consumerId={}",
                      remoteAddress, consumerId);
+            consumers.remove(consumerId, consumerFuture);
             commandSender.sendSuccessResponse(requestId);
             return;
         }

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java

@@ -2436,44 +2436,30 @@ public void testSubscribeBookieTimeout() throws Exception {
                 null /* assignedBrokerServiceUrl */, null /* assignedBrokerServiceUrlTls */);
         channel.writeInbound(closeConsumer);
 
-        ByteBuf subscribe2 = Commands.newSubscribe(successTopicName, //
-                successSubName, 1 /* consumer id */, 3 /* request id */, SubType.Exclusive, 0,
-                "test" /* consumer name */, 0 /* avoid reseting cursor */);
-        channel.writeInbound(subscribe2);
-
-        openTopicFail.get().run();
-
         Object response;
 
         // Close succeeds
         response = getResponse();
         assertEquals(response.getClass(), CommandSuccess.class);
         assertEquals(((CommandSuccess) response).getRequestId(), 2);
 
-        // Subscribe fails
-        response = getResponse();
-        assertEquals(response.getClass(), CommandError.class);
-        assertEquals(((CommandError) response).getRequestId(), 3);
+        // We should not receive response for 1st consumer, since it was cancelled by the close
+        openTopicFail.get().run();
 
-        Awaitility.await().until(() -> !serverCnx.hasConsumer(1));
+        assertTrue(channel.outboundMessages().isEmpty());
+        assertFalse(serverCnx.hasConsumer(1));
 
-        ByteBuf subscribe3 = Commands.newSubscribe(successTopicName, //
-                successSubName, 1 /* consumer id */, 4 /* request id */, SubType.Exclusive, 0,
+        ByteBuf subscribe2 = Commands.newSubscribe(successTopicName, //
+                successSubName, 1 /* consumer id */, 3 /* request id */, SubType.Exclusive, 0,
                 "test" /* consumer name */, 0 /* avoid reseting cursor */);
-        channel.writeInbound(subscribe3);
+        channel.writeInbound(subscribe2);
 
         openTopicSuccess.get().run();
 
         // Subscribe succeeds
         response = getResponse();
         assertEquals(response.getClass(), CommandSuccess.class);
-        assertEquals(((CommandSuccess) response).getRequestId(), 4);
-
-        Thread.sleep(100);
-
-        // We should not receive response for 1st producer, since it was cancelled by the close
-        assertTrue(channel.outboundMessages().isEmpty());
-        assertTrue(channel.isActive());
+        assertEquals(((CommandSuccess) response).getRequestId(), 3);
 
         channel.finish();
     }