diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/AbstractSystemMessageSyncer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/AbstractSystemMessageSyncer.java index 05eb6726188..5e0d82f9822 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/AbstractSystemMessageSyncer.java +++ b/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"; } } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java index 9a2c5e3437d..cff69bdf583 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java +++ b/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 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()); } } -} \ No newline at end of file +}