fix: summarize heartbeat and system-message sync logs

proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/AbstractSystemMessageSyncer.java

@@ -93,6 +93,7 @@ public RPCHook getRpcHook() {
 
     protected void sendSystemMessage(Object data) {
         String targetTopic = this.getBroadcastTopicName();
+        String dataSummary = safelySummarizeSystemMessageData(data);
         try {
             Message message = new Message(
                 targetTopic,
@@ -109,15 +110,43 @@ protected void sendSystemMessage(Object data) {
                 Duration.ofSeconds(3).toMillis()
             ).whenCompleteAsync((result, throwable) -> {
                 if (throwable != null) {
-                    log.error("send system message failed. data: {}, topic: {}", data, getBroadcastTopicName(), throwable);
+                    log.error("send system message failed. dataSummary: {}, topic: {}",
+                        dataSummary, targetTopic, throwable);
                     return;
                 }
                 if (SendStatus.SEND_OK != result.getSendStatus()) {
-                    log.error("send system message failed. data: {}, topic: {}, sendResult:{}", data, getBroadcastTopicName(), result);
+                    log.error("send system message failed. dataSummary: {}, topic: {}, sendResult:{}",
+                        dataSummary, targetTopic, result);
                 }
             });
         } catch (Throwable t) {
-            log.error("send system message failed. data: {}, topic: {}", data, targetTopic, t);
+            log.error("send system message failed. dataSummary: {}, topic: {}", dataSummary, targetTopic, t);
+        }
+    }
+
+    static String summarizeSystemMessageData(Object data) {
+        if (data == null) {
+            return "null";
+        }
+        if (data instanceof HeartbeatSyncerData) {
+            HeartbeatSyncerData heartbeatData = (HeartbeatSyncerData) data;
+            int subscriptionCount = heartbeatData.getSubscriptionDataSet() == null
+                ? 0 : heartbeatData.getSubscriptionDataSet().size();
+            return "HeartbeatSyncerData{"
+                + "heartbeatType=" + heartbeatData.getHeartbeatType()
+                + ", subscriptionCount=" + subscriptionCount
+                + ", channelDataPresent=" + (heartbeatData.getChannelData() != null)
+                + '}';
+        }
+        String simpleName = data.getClass().getSimpleName();
+        return StringUtils.isEmpty(simpleName) ? data.getClass().getName() : simpleName;
+    }
+
+    static String safelySummarizeSystemMessageData(Object data) {
+        try {
+            return summarizeSystemMessageData(data);
+        } catch (Throwable ignored) {
+            return "unavailable";
         }
     }
 

proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncer.java

@@ -18,6 +18,7 @@
 package org.apache.rocketmq.proxy.service.sysmessage;
 
 import com.alibaba.fastjson2.JSON;
+import com.google.common.base.MoreObjects;
 import io.netty.channel.Channel;
 import org.apache.rocketmq.broker.client.ClientChannelInfo;
 import org.apache.rocketmq.broker.client.ConsumerGroupEvent;
@@ -47,6 +48,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 public class HeartbeatSyncer extends AbstractSystemMessageSyncer {
 
@@ -131,16 +133,19 @@ public void onConsumerRegister(String consumerGroup, ClientChannelInfo clientCha
                     );
                     data.setSubscriptionDataSet(subList);
 
-                    log.debug("sync register heart beat. topic:{}, data:{}", this.getBroadcastTopicName(), data);
+                    log.debug("sync register heart beat. topic:{}, dataSummary:{}",
+                        this.getBroadcastTopicName(), summarizeHeartbeatData(data));
                     this.sendSystemMessage(data);
                 } catch (Throwable t) {
-                    log.error("heartbeat register broadcast failed. group:{}, clientChannelInfo:{}, consumeType:{}, messageModel:{}, consumeFromWhere:{}, subList:{}",
-                        consumerGroup, clientChannelInfo, consumeType, messageModel, consumeFromWhere, subList, t);
+                    log.error("heartbeat register broadcast failed. group:{}, clientChannelInfo:{}, consumeType:{}, messageModel:{}, consumeFromWhere:{}, subscriptionSummary:{}",
+                        consumerGroup, clientChannelInfo, consumeType, messageModel, consumeFromWhere,
+                        summarizeSubscriptionDataSet(subList), t);
                 }
             });
         } catch (Throwable t) {
-            log.error("heartbeat submit register broadcast failed. group:{}, clientChannelInfo:{}, consumeType:{}, messageModel:{}, consumeFromWhere:{}, subList:{}",
-                consumerGroup, clientChannelInfo, consumeType, messageModel, consumeFromWhere, subList, t);
+            log.error("heartbeat submit register broadcast failed. group:{}, clientChannelInfo:{}, consumeType:{}, messageModel:{}, consumeFromWhere:{}, subscriptionSummary:{}",
+                consumerGroup, clientChannelInfo, consumeType, messageModel, consumeFromWhere,
+                summarizeSubscriptionDataSet(subList), t);
         }
     }
 
@@ -168,7 +173,8 @@ public void onConsumerUnRegister(String consumerGroup, ClientChannelInfo clientC
                         remoteChannel.encode()
                     );
 
-                    log.debug("sync unregister heart beat. topic:{}, data:{}", this.getBroadcastTopicName(), data);
+                    log.debug("sync unregister heart beat. topic:{}, dataSummary:{}",
+                        this.getBroadcastTopicName(), summarizeHeartbeatData(data));
                     this.sendSystemMessage(data);
                 } catch (Throwable t) {
                     log.error("heartbeat unregister broadcast failed. group:{}, clientChannelInfo:{}, consumeType:{}",
@@ -188,8 +194,9 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeCo
         }
 
         for (MessageExt msg : msgs) {
+            HeartbeatSyncerData data = null;
             try {
-                HeartbeatSyncerData data = JSON.parseObject(new String(msg.getBody(), StandardCharsets.UTF_8), HeartbeatSyncerData.class);
+                data = JSON.parseObject(new String(msg.getBody(), StandardCharsets.UTF_8), HeartbeatSyncerData.class);
                 if (data.getLocalProxyId().equals(localProxyId)) {
                     continue;
                 }
@@ -203,7 +210,8 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeCo
                     data.getLanguage(),
                     data.getVersion()
                 );
-                log.debug("start process remote channel. data:{}, clientChannelInfo:{}", data, clientChannelInfo);
+                log.debug("start process remote channel. dataSummary:{}, clientChannelInfo:{}",
+                    summarizeHeartbeatData(data), clientChannelInfo);
                 if (data.getHeartbeatType().equals(HeartbeatType.REGISTER)) {
                     this.consumerManager.registerConsumer(
                         data.getGroup(),
@@ -222,13 +230,31 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeCo
                     );
                 }
             } catch (Throwable t) {
-                log.error("heartbeat consume message failed. msg:{}, data:{}", msg, new String(msg.getBody(), StandardCharsets.UTF_8), t);
+                log.error("heartbeat consume message failed. summary:{}", summarizeHeartbeatMessage(msg, data), t);
             }
         }
 
         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
     }
 
+    static String summarizeHeartbeatMessage(MessageExt msg, HeartbeatSyncerData data) {
+        if (msg == null) {
+            return "msg=null";
+        }
+        StringBuilder summary = new StringBuilder()
+            .append("topic=").append(msg.getTopic())
+            .append(", msgId=").append(msg.getMsgId())
+            .append(", bodySize=").append(msg.getBody() == null ? 0 : msg.getBody().length);
+        if (data != null) {
+            summary.append(", heartbeatType=").append(data.getHeartbeatType())
+                .append(", group=").append(data.getGroup())
+                .append(", clientId=").append(data.getClientId())
+                .append(", subscriptionCount=")
+                .append(data.getSubscriptionDataSet() == null ? 0 : data.getSubscriptionDataSet().size());
+        }
+        return summary.toString();
+    }
+
     private String buildLocalProxyId() {
         ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
         // use local address, remoting port and grpc port to build unique local proxy Id
@@ -238,4 +264,50 @@ private String buildLocalProxyId() {
     private static String buildKey(String group, Channel channel) {
         return group + "@" + channel.id().asLongText();
     }
+
+    static String summarizeHeartbeatData(HeartbeatSyncerData data) {
+        if (data == null) {
+            return "null";
+        }
+        return MoreObjects.toStringHelper("HeartbeatSyncerData")
+            .add("heartbeatType", data.getHeartbeatType())
+            .add("clientId", data.getClientId())
+            .add("language", data.getLanguage())
+            .add("version", data.getVersion())
+            .add("group", data.getGroup())
+            .add("consumeType", data.getConsumeType())
+            .add("messageModel", data.getMessageModel())
+            .add("consumeFromWhere", data.getConsumeFromWhere())
+            .add("localProxyId", data.getLocalProxyId())
+            .add("channelDataPresent", data.getChannelData() != null)
+            .add("subscriptionSummary", summarizeSubscriptionDataSet(data.getSubscriptionDataSet()))
+            .toString();
+    }
+
+    static String summarizeSubscriptionDataSet(Set<SubscriptionData> subscriptions) {
+        if (subscriptions == null) {
+            return "null";
+        }
+        List<String> topics = subscriptions.stream()
+            .map(SubscriptionData::getTopic)
+            .sorted()
+            .collect(Collectors.toList());
+        return MoreObjects.toStringHelper("SubscriptionDataSet")
+            .add("count", subscriptions.size())
+            .add("topics", topics)
+            .toString();
+    }
+
+    static String summarizeSystemMessage(MessageExt msg) {
+        if (msg == null) {
+            return "null";
+        }
+        byte[] body = msg.getBody();
+        return MoreObjects.toStringHelper("MessageExt")
+            .add("topic", msg.getTopic())
+            .add("msgId", msg.getMsgId())
+            .add("bodyBytes", body == null ? 0 : body.length)
+            .toString();
+    }
+
 }

proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java

@@ -23,9 +23,11 @@
 import apache.rocketmq.v2.Settings;
 import apache.rocketmq.v2.Subscription;
 import apache.rocketmq.v2.SubscriptionEntry;
+import com.alibaba.fastjson2.JSON;
 import com.google.common.collect.Sets;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelId;
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
@@ -77,6 +79,7 @@
 
 import static org.awaitility.Awaitility.await;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
@@ -217,6 +220,88 @@ public void testSyncGrpcV2Channel() throws Exception {
         assertSame(channelInfoList.get(0).getChannel(), syncUnRegisterChannelInfoArgumentCaptor.getValue().getChannel());
     }
 
+    @Test
+    public void testSummarizeHeartbeatDataDoesNotExposeSubscriptionExpressions() throws Exception {
+        String expression = "secretTagA || secretTagB";
+        HeartbeatSyncerData data = new HeartbeatSyncerData(
+            HeartbeatType.REGISTER,
+            clientId,
+            LanguageCode.JAVA,
+            5,
+            "consumerGroup",
+            ConsumeType.CONSUME_PASSIVELY,
+            MessageModel.CLUSTERING,
+            ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET,
+            "proxy-0",
+            "raw-channel-data"
+        );
+        data.setSubscriptionDataSet(Sets.newHashSet(
+            FilterAPI.buildSubscriptionData("topic-a", expression),
+            FilterAPI.buildSubscriptionData("topic-b", "*")
+        ));
+
+        String summary = HeartbeatSyncer.summarizeHeartbeatData(data);
+
+        assertTrue(summary.contains("heartbeatType=REGISTER"));
+        assertTrue(summary.contains("clientId=" + clientId));
+        assertTrue(summary.contains("group=consumerGroup"));
+        assertTrue(summary.contains("channelDataPresent=true"));
+        assertTrue(summary.contains("count=2"));
+        assertTrue(summary.contains("topic-a"));
+        assertTrue(summary.contains("topic-b"));
+        assertFalse(summary.contains(expression));
+        assertFalse(summary.contains("raw-channel-data"));
+    }
+
+    @Test
+    public void testSummarizeSystemMessageDataAvoidsDetailedHeartbeatPayload() throws Exception {
+        HeartbeatSyncerData data = new HeartbeatSyncerData(
+            HeartbeatType.REGISTER,
+            clientId,
+            LanguageCode.JAVA,
+            5,
+            "sensitiveGroup",
+            ConsumeType.CONSUME_PASSIVELY,
+            MessageModel.CLUSTERING,
+            ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET,
+            "localProxyId",
+            "sensitiveChannelData"
+        );
+        data.setSubscriptionDataSet(Sets.newHashSet(FilterAPI.buildSubscriptionData("sensitiveTopic", "sensitiveTag")));
+
+        String summary = AbstractSystemMessageSyncer.summarizeSystemMessageData(data);
+
+        assertTrue(summary.contains("HeartbeatSyncerData"));
+        assertTrue(summary.contains("heartbeatType=REGISTER"));
+        assertTrue(summary.contains("subscriptionCount=1"));
+        assertTrue(summary.contains("channelDataPresent=true"));
+        assertFalse(summary.contains(clientId));
+        assertFalse(summary.contains("sensitiveGroup"));
+        assertFalse(summary.contains("sensitiveTopic"));
+        assertFalse(summary.contains("sensitiveTag"));
+        assertFalse(summary.contains("sensitiveChannelData"));
+    }
+
+    @Test
+    public void testSummarizeSystemMessageDataUsesClassNameForAnonymousPayload() {
+        Object data = new Object() {
+        };
+
+        assertEquals(data.getClass().getName(), AbstractSystemMessageSyncer.summarizeSystemMessageData(data));
+    }
+
+    @Test
+    public void testSafelySummarizeSystemMessageDataDoesNotPropagateFailures() {
+        HeartbeatSyncerData data = new HeartbeatSyncerData() {
+            @Override
+            public Set<SubscriptionData> getSubscriptionDataSet() {
+                throw new IllegalStateException("summary failure");
+            }
+        };
+
+        assertEquals("unavailable", AbstractSystemMessageSyncer.safelySummarizeSystemMessageData(data));
+    }
+
     @Test
     public void testSyncRemotingChannel() throws Exception {
         String consumerGroup = "consumerGroup";
@@ -388,6 +473,53 @@ private void testProcessConsumerGroupEvent(String consumerGroup, ClientChannelIn
         assertTrue(heartbeatSyncer.remoteChannelMap.isEmpty());
     }
 
+    @Test
+    public void testSummarizeHeartbeatMessageDoesNotExposeSubscriptionOrChannelData() throws Exception {
+        SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData("topic", "secret-tag");
+        HeartbeatSyncerData data = new HeartbeatSyncerData(
+            HeartbeatType.REGISTER,
+            "client-secret",
+            LanguageCode.JAVA,
+            5,
+            "consumerGroup",
+            ConsumeType.CONSUME_PASSIVELY,
+            MessageModel.CLUSTERING,
+            ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET,
+            "proxyId",
+            "secret-channel-data"
+        );
+        data.setSubscriptionDataSet(Sets.newHashSet(subscriptionData));
+        MessageExt msg = new MessageExt();
+        msg.setTopic("heartbeatTopic");
+        msg.setMsgId("msgId");
+        msg.setBody(JSON.toJSONString(data).getBytes(StandardCharsets.UTF_8));
+
+        String summary = HeartbeatSyncer.summarizeHeartbeatMessage(msg, data);
+
+        assertTrue(summary.contains("topic=heartbeatTopic"));
+        assertTrue(summary.contains("msgId=msgId"));
+        assertTrue(summary.contains("heartbeatType=REGISTER"));
+        assertTrue(summary.contains("group=consumerGroup"));
+        assertTrue(summary.contains("subscriptionCount=1"));
+        assertFalse(summary.contains("secret-tag"));
+        assertFalse(summary.contains("secret-channel-data"));
+    }
+
+    @Test
+    public void testSummarizeHeartbeatMessageDoesNotExposeUnparsedBody() {
+        MessageExt msg = new MessageExt();
+        msg.setTopic("heartbeatTopic");
+        msg.setMsgId("msgId");
+        msg.setBody("raw-secret-body".getBytes(StandardCharsets.UTF_8));
+
+        String summary = HeartbeatSyncer.summarizeHeartbeatMessage(msg, null);
+
+        assertTrue(summary.contains("topic=heartbeatTopic"));
+        assertTrue(summary.contains("msgId=msgId"));
+        assertTrue(summary.contains("bodySize=15"));
+        assertFalse(summary.contains("raw-secret-body"));
+    }
+
     private MessageExt convertFromMessage(Message message) {
         MessageExt messageExt = new MessageExt();
         messageExt.setTopic(message.getTopic());
@@ -433,4 +565,4 @@ public int compareTo(@NotNull ChannelId o) {
             return this.channelId.compareTo(o.asLongText());
         }
     }
-}
+}