[ISSUE #10760] Avoid logging full system message data

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/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java

@@ -77,6 +77,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 +218,55 @@ public void testSyncGrpcV2Channel() throws Exception {
         assertSame(channelInfoList.get(0).getChannel(), syncUnRegisterChannelInfoArgumentCaptor.getValue().getChannel());
     }
 
+    @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";
@@ -433,4 +483,4 @@ public int compareTo(@NotNull ChannelId o) {
             return this.channelId.compareTo(o.asLongText());
         }
     }
-}
+}