[ISSUE #8970] Remove redundant heartbeats

client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java

@@ -16,6 +16,7 @@
  */
 package org.apache.rocketmq.client.impl.factory;
 
+import com.alibaba.fastjson.JSON;
 import io.netty.channel.Channel;
 import java.util.Collections;
 import java.util.HashMap;
@@ -35,7 +36,6 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
-import com.alibaba.fastjson.JSON;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.client.ClientConfig;
 import org.apache.rocketmq.client.admin.MQAdminExtInner;
@@ -66,6 +66,8 @@
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.common.message.MessageQueueAssignment;
 import org.apache.rocketmq.common.topic.TopicValidator;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 import org.apache.rocketmq.remoting.ChannelEventListener;
 import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.remoting.common.HeartbeatV2Result;
@@ -83,8 +85,6 @@
 import org.apache.rocketmq.remoting.protocol.route.BrokerData;
 import org.apache.rocketmq.remoting.protocol.route.QueueData;
 import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
-import org.apache.rocketmq.logging.org.slf4j.Logger;
-import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 
 import static org.apache.rocketmq.remoting.rpc.ClientMetadata.topicRouteData2EndpointsForStaticTopic;
 
@@ -157,7 +157,9 @@ public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String cli
         ChannelEventListener channelEventListener;
         if (clientConfig.isEnableHeartbeatChannelEventListener()) {
             channelEventListener = new ChannelEventListener() {
+
                 private final ConcurrentMap<String, HashMap<Long, String>> brokerAddrTable = MQClientInstance.this.brokerAddrTable;
+
                 @Override
                 public void onChannelConnect(String remoteAddr, Channel channel) {
                 }
@@ -182,7 +184,7 @@ public void onChannelActive(String remoteAddr, Channel channel) {
                             if (addr.equals(remoteAddr)) {
                                 long id = entry.getKey();
                                 String brokerName = addressEntry.getKey();
-                                if (sendHeartbeatToBroker(id, brokerName, addr)) {
+                                if (sendHeartbeatToBroker(id, brokerName, addr, false)) {
                                     rebalanceImmediately();
                                 }
                                 break;
@@ -591,6 +593,18 @@ private boolean isBrokerAddrExistInTopicRouteTable(final String addr) {
     }
 
     public boolean sendHeartbeatToBroker(long id, String brokerName, String addr) {
+        return sendHeartbeatToBroker(id, brokerName, addr, true);
+    }
+
+    /**
+     * @param id
+     * @param brokerName
+     * @param addr
+     * @param strictLockMode When the connection is initially established, sending a heartbeat will simultaneously trigger the onChannelActive event to acquire the lock again, causing an exception. Therefore,
+     *                       the exception that occurs when sending the heartbeat during the initial onChannelActive event can be ignored.
+     * @return
+     */
+    public boolean sendHeartbeatToBroker(long id, String brokerName, String addr, boolean strictLockMode) {
         if (this.lockHeartbeat.tryLock()) {
             final HeartbeatData heartbeatDataWithSub = this.prepareHeartbeatData(false);
             final boolean producerEmpty = heartbeatDataWithSub.getProducerDataSet().isEmpty();
@@ -615,7 +629,9 @@ public boolean sendHeartbeatToBroker(long id, String brokerName, String addr) {
                 this.lockHeartbeat.unlock();
             }
         } else {
-            log.warn("lock heartBeat, but failed. [{}]", this.clientId);
+            if (strictLockMode) {
+                log.warn("lock heartBeat, but failed. [{}]", this.clientId);
+            }
         }
         return false;
     }