[ISSUE #9725] Maintain heartbeat between Proxy and Broker

client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java

@@ -18,6 +18,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -149,6 +150,26 @@ public CompletableFuture<Void> sendHeartbeatOneway(
         return future;
     }
 
+    /**
+     * Return active broker channels while excluding configured and discovered NameServer channels.
+     */
+    public Set<String> getActiveBrokerAddresses() {
+        List<String> activeChannelAddresses = this.getRemotingClient().getActiveChannelAddresses();
+        Set<String> activeAddresses = activeChannelAddresses == null
+            ? new HashSet<>() : new HashSet<>(activeChannelAddresses);
+        List<String> configuredNameServers = this.getRemotingClient().getNameServerAddressList();
+        if (configuredNameServers != null) {
+            activeAddresses.removeAll(configuredNameServers);
+        }
+        List<String> availableNameServers = this.getRemotingClient().getAvailableNameSrvList();
+        if (availableNameServers != null) {
+            activeAddresses.removeAll(availableNameServers);
+        }
+        activeAddresses.remove(null);
+        activeAddresses.remove("");
+        return activeAddresses;
+    }
+
     public CompletableFuture<Integer> sendHeartbeatAsync(
         String brokerAddr,
         HeartbeatData heartbeatData,

proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java

@@ -164,6 +164,13 @@ public class ProxyConfig implements ConfigFile {
 
     private int rocketmqMQClientNum = 6;
 
+    /**
+     * Keep already established Proxy-to-Broker remoting channels alive during low-traffic periods.
+     */
+    private boolean enableProxyBrokerHeartbeat = true;
+    private long proxyBrokerHeartbeatIntervalMillis = Duration.ofSeconds(30).toMillis();
+    private long proxyBrokerHeartbeatTimeoutMillis = Duration.ofSeconds(3).toMillis();
+
     private long grpcProxyRelayRequestTimeoutInSeconds = 5;
     private int grpcProducerThreadPoolNums = PROCESSOR_NUMBER;
     private int grpcProducerThreadQueueCapacity = 10000;
@@ -296,6 +303,14 @@ public class ProxyConfig implements ConfigFile {
     @Override
     public void initData() {
         parseDelayLevel();
+        if (enableProxyBrokerHeartbeat && proxyBrokerHeartbeatIntervalMillis <= 0) {
+            throw new ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR,
+                "proxyBrokerHeartbeatIntervalMillis must be greater than zero");
+        }
+        if (enableProxyBrokerHeartbeat && proxyBrokerHeartbeatTimeoutMillis <= 0) {
+            throw new ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR,
+                "proxyBrokerHeartbeatTimeoutMillis must be greater than zero");
+        }
         if (StringUtils.isEmpty(localServeAddr)) {
             this.localServeAddr = NetworkUtil.getLocalAddress();
         }
@@ -739,6 +754,30 @@ public void setRocketmqMQClientNum(int rocketmqMQClientNum) {
         this.rocketmqMQClientNum = rocketmqMQClientNum;
     }
 
+    public boolean isEnableProxyBrokerHeartbeat() {
+        return enableProxyBrokerHeartbeat;
+    }
+
+    public void setEnableProxyBrokerHeartbeat(boolean enableProxyBrokerHeartbeat) {
+        this.enableProxyBrokerHeartbeat = enableProxyBrokerHeartbeat;
+    }
+
+    public long getProxyBrokerHeartbeatIntervalMillis() {
+        return proxyBrokerHeartbeatIntervalMillis;
+    }
+
+    public void setProxyBrokerHeartbeatIntervalMillis(long proxyBrokerHeartbeatIntervalMillis) {
+        this.proxyBrokerHeartbeatIntervalMillis = proxyBrokerHeartbeatIntervalMillis;
+    }
+
+    public long getProxyBrokerHeartbeatTimeoutMillis() {
+        return proxyBrokerHeartbeatTimeoutMillis;
+    }
+
+    public void setProxyBrokerHeartbeatTimeoutMillis(long proxyBrokerHeartbeatTimeoutMillis) {
+        this.proxyBrokerHeartbeatTimeoutMillis = proxyBrokerHeartbeatTimeoutMillis;
+    }
+
     public long getGrpcProxyRelayRequestTimeoutInSeconds() {
         return grpcProxyRelayRequestTimeoutInSeconds;
     }

proxy/src/main/java/org/apache/rocketmq/proxy/service/ClusterServiceManager.java

@@ -16,6 +16,7 @@
  */
 package org.apache.rocketmq.proxy.service;
 
+import java.util.Arrays;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.broker.client.ClientChannelInfo;
@@ -73,6 +74,7 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S
     protected MQClientAPIFactory operationClientAPIFactory;
     protected MQClientAPIFactory transactionClientAPIFactory;
     protected MQClientAPIFactory liteSubscriptionAPIFactory;
+    protected ProxyBrokerHeartbeatService proxyBrokerHeartbeatService;
 
     public ClusterServiceManager(RPCHook rpcHook) {
         this(rpcHook, null);
@@ -136,6 +138,11 @@ public ClusterServiceManager(RPCHook rpcHook, ObjectCreator<RemotingClient> remo
             scheduledExecutorService);
         this.liteSubscriptionService = new LiteSubscriptionService(this.topicRouteService, this.liteSubscriptionAPIFactory);
 
+        this.proxyBrokerHeartbeatService = new ProxyBrokerHeartbeatService(
+            Arrays.asList(this.messagingClientAPIFactory, this.operationClientAPIFactory,
+                this.transactionClientAPIFactory, this.liteSubscriptionAPIFactory),
+            this.scheduledExecutorService, proxyConfig);
+
         this.init();
     }
 
@@ -156,6 +163,7 @@ protected void init() {
         this.appendStartAndShutdown(this.operationClientAPIFactory);
         this.appendStartAndShutdown(this.transactionClientAPIFactory);
         this.appendStartAndShutdown(this.liteSubscriptionAPIFactory);
+        this.appendStartAndShutdown(this.proxyBrokerHeartbeatService);
         this.appendStartAndShutdown(this.topicRouteService);
         this.appendStartAndShutdown(this.clusterTransactionService);
         this.appendStartAndShutdown(this.metadataService);