[fix][broker]A failed consumer/producer future in ServerCnx can never be removed

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

@@ -3605,19 +3605,54 @@ public String clientSourceAddressAndPort() {
 
     @Override
     public CompletableFuture<Optional<Boolean>> checkConnectionLiveness() {
+        if (!isActive()) {
+            return CompletableFuture.completedFuture(Optional.of(false));
+        }
         if (connectionLivenessCheckTimeoutMillis > 0) {
             return NettyFutureUtil.toCompletableFuture(ctx.executor().submit(() -> {
+                if (!isActive()) {
+                    return CompletableFuture.completedFuture(Optional.of(false));
+                }
                 if (connectionCheckInProgress != null) {
                     return connectionCheckInProgress;
                 } else {
                     final CompletableFuture<Optional<Boolean>> finalConnectionCheckInProgress =
                             new CompletableFuture<>();
                     connectionCheckInProgress = finalConnectionCheckInProgress;
                     ctx.executor().schedule(() -> {
-                        if (finalConnectionCheckInProgress == connectionCheckInProgress
-                                && !finalConnectionCheckInProgress.isDone()) {
+                        if (!isActive()) {
+                            finalConnectionCheckInProgress.complete(Optional.of(false));
+                            return;
+                        }
+                        if (finalConnectionCheckInProgress.isDone()) {
+                            return;
+                        }
+                        if (finalConnectionCheckInProgress == connectionCheckInProgress) {
+                            /**
+                             * {@link #connectionCheckInProgress} will be completed when
+                             * {@link #channelInactive(ChannelHandlerContext)} event occurs, so skip set it here.
+                             */
                             log.warn("[{}] Connection check timed out. Closing connection.", this.toString());
                             ctx.close();
+                        } else {
+                            /**
+                             * Scenarios that changing {@link #connectionCheckInProgress}.
+                             *   1. {@link #connectionCheckInProgress} will be changed to "null" after a successful
+                             *     "Ping & Pong"
+                             *   2. {@link #connectionCheckInProgress} will be set to a new future the next time
+                             *     {@link #checkConnectionLiveness()} when it is null.
+                             *
+                             * Once {@link #connectionCheckInProgress} is not equal to
+                             *   {@link #finalConnectionCheckInProgress}, it means Scenario 1 occurred before. If
+                             *   "receiving Pong" and "the current scheduled task" are executing at the same time,
+                             *   this log will be printed. Since the two events will be executing at the same thread,
+                             *   the concurrency scenario can not occur, so this log's level is "ERROR".
+                             */
+                            log.error("[{}] Connection check might be success, because the variable"
+                                    + " connectionCheckInProgress has been override by the following check. But this"
+                                    + " scenario is not expected",
+                                    this.toString());
+                            finalConnectionCheckInProgress.complete(Optional.of(true));
                         }
                     }, connectionLivenessCheckTimeoutMillis, TimeUnit.MILLISECONDS);
                     sendPing();

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

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.broker.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class ServerCnxNonInjectionTest extends ProducerConsumerBase {
+
+    @BeforeClass
+    @Override
+    protected void setup() throws Exception {
+        super.internalSetup();
+        super.producerBaseSetup();
+    }
+
+    @AfterClass
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @Test(timeOut = 60 * 1000)
+    public void testCheckConnectionLivenessAfterClosed() throws Exception {
+        // Create a ServerCnx
+        final String tp = BrokerTestUtil.newUniqueName("public/default/tp");
+        Producer<String> p = pulsarClient.newProducer(Schema.STRING).topic(tp).create();
+        ServerCnx serverCnx = (ServerCnx) pulsar.getBrokerService().getTopic(tp, false).join().get()
+                        .getProducers().values().iterator().next().getCnx();
+        // Call "CheckConnectionLiveness" after serverCnx is closed. The resulted future should be done eventually.
+        p.close();
+        serverCnx.close();
+        Thread.sleep(1000);
+        serverCnx.checkConnectionLiveness().join();
+    }
+
+}