diff --git a/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java index c595178d193..d77888283bb 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java @@ -180,6 +180,12 @@ public void notifyMessageArrivingWithRetryTopic(final String topic, final int qu private void notifyMessageArrivingFromRetry(String topic, int queueId, Long tagsCode, long msgStoreTime, byte[] filterBitMap, Map properties) { + if (properties == null) { + // A retry topic message without properties can't be mapped back to its origin group, + // so there is no long polling request to wake up. Throwing here would poison the + // reput thread for every following message. + return; + } String prefix = MixAll.RETRY_GROUP_TOPIC_PREFIX; String originGroup = properties.get(MessageConst.PROPERTY_ORIGIN_GROUP); // In the case of pop consumption, there is no long polling hanging on the retry topic, so the wake-up is skipped. diff --git a/broker/src/test/java/org/apache/rocketmq/broker/longpolling/PopLongPollingServiceTest.java b/broker/src/test/java/org/apache/rocketmq/broker/longpolling/PopLongPollingServiceTest.java index 23dcf5c2fda..60b977d522f 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/longpolling/PopLongPollingServiceTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/longpolling/PopLongPollingServiceTest.java @@ -47,8 +47,13 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -110,6 +115,19 @@ public void testNotifyMessageArrivingFromRetry() { verify(popLongPollingService, times(2)).notifyMessageArriving(defaultTopic, queueId, group, true, -1L, 0L, null, properties, null); } + @Test + public void testNotifyMessageArrivingFromRetryWithoutProperties() { + int queueId = -1; + String group = "group"; + String pullRetryTopic = MixAll.getRetryTopic(group); + // A message stored on a retry topic without any properties (properties map is null, + // e.g. written by a non-Java client) must not throw, otherwise the reput thread + // stalls dispatch of every following message. + popLongPollingService.notifyMessageArrivingWithRetryTopic(pullRetryTopic, queueId, queueId, -1L, 0L, null, null); + verify(popLongPollingService, never()).notifyMessageArriving(anyString(), anyInt(), anyString(), anyBoolean(), + any(), anyLong(), any(), any(), any()); + } + @Test public void testNotifyMessageArriving() { int queueId = 0;