From 8f6a9c9b459b689c3bf5d129a7bc76ab94d29254 Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 3 Jul 2026 11:47:54 +0800 Subject: [PATCH] Avoid receive failures when client settings are absent ReceiveMessageActivity can be invoked before cached gRPC client settings are available. When settings are missing, fall back to the real default consumer settings (maxAttempts = retryMaxTimes + 1) instead of Settings.getDefaultInstance(), whose backoffPolicy.maxAttempts is 0. With maxAttempts == 0, PopMessageResultFilterImpl routes fresh messages (reconsumeTimes == 0) straight to the DLQ via reconsumeTimes >= maxAttempts. The real default avoids both the NullPointerException and the DLQ regression. Constraint: Related to apache/rocketmq#8714 Confidence: high Scope-risk: narrow Directive: Keep missing client settings on the default consumer path unless the protocol starts requiring an explicit settings handshake before receive. Tested: mvn -pl proxy -Dtest=ReceiveMessageActivityTest,PopMessageResultFilterImplTest,GrpcClientSettingsManagerTest test Not-tested: Full repository test suite --- .../v2/common/GrpcClientSettingsManager.java | 12 +++ .../v2/consumer/ReceiveMessageActivity.java | 3 + .../common/GrpcClientSettingsManagerTest.java | 15 ++++ .../PopMessageResultFilterImplTest.java | 90 +++++++++++++++++++ .../consumer/ReceiveMessageActivityTest.java | 59 ++++++++++++ 5 files changed, 179 insertions(+) create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/PopMessageResultFilterImplTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java index 75cac21be4a..6bd8571b392 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java @@ -197,6 +197,18 @@ protected Settings.Builder createDefaultConsumerSettingsBuilder() { .toBuilder(); } + /** + * Build the default consumer settings derived from the default {@link SubscriptionGroupConfig}. + * + * Unlike {@link Settings#getDefaultInstance()}, which leaves {@code backoffPolicy.maxAttempts} at the protobuf + * default of 0, this yields a real consumer default (e.g. {@code maxAttempts = retryMaxTimes + 1}) so that + * {@link org.apache.rocketmq.proxy.grpc.v2.consumer.PopMessageResultFilterImpl} does not route fresh messages + * ({@code reconsumeTimes == 0}) straight to the DLQ when client settings are not yet cached. + */ + public Settings getDefaultConsumerSettings() { + return createDefaultConsumerSettingsBuilder().build(); + } + public Settings removeAndGetRawClientSettings(String clientId) { return CLIENT_SETTINGS_MAP.remove(clientId); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java index becf2c2165d..571af8c1c53 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java @@ -65,6 +65,9 @@ public void receiveMessage(ProxyContext ctx, ReceiveMessageRequest request, try { Settings settings = this.grpcClientSettingsManager.getClientSettings(ctx); + if (settings == null) { + settings = this.grpcClientSettingsManager.getDefaultConsumerSettings(); + } final boolean isLite = ClientType.LITE_PUSH_CONSUMER.equals(settings.getClientType()); Subscription subscription = settings.getSubscription(); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java index 4d0037a272a..c99ef9ee2bb 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java @@ -42,6 +42,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; @@ -76,6 +77,20 @@ public void testGetProducerData() { assertNotEquals(settings.getPublishing(), settings.getPublishing().getDefaultInstanceForType()); } + @Test + public void testDefaultConsumerSettingsHasNonZeroMaxAttempts() { + // When client settings are missing, ReceiveMessageActivity falls back to the default consumer settings. + // The protobuf empty default leaves backoffPolicy.maxAttempts at 0, which makes + // PopMessageResultFilterImpl route fresh messages (reconsumeTimes == 0) straight to the DLQ. + // The real consumer default must carry a positive maxAttempts to avoid that. + Settings defaultSettings = this.grpcClientSettingsManager.getDefaultConsumerSettings(); + int maxAttempts = defaultSettings.getBackoffPolicy().getMaxAttempts(); + assertNotEquals("default consumer settings must not reuse the protobuf empty default", + RetryPolicy.getDefaultInstance(), defaultSettings.getBackoffPolicy()); + assertTrue("default consumer maxAttempts must be positive so fresh messages are not DLQ'd, got " + maxAttempts, + maxAttempts > 0); + } + @Test public void testGetSubscriptionData() { ProxyContext context = ProxyContext.create().withVal(ContextVariable.CLIENT_ID, CLIENT_ID); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/PopMessageResultFilterImplTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/PopMessageResultFilterImplTest.java new file mode 100644 index 00000000000..b3e3fd96b91 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/PopMessageResultFilterImplTest.java @@ -0,0 +1,90 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.rocketmq.proxy.grpc.v2.consumer; + +import java.util.Collections; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest; +import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; +import org.apache.rocketmq.proxy.processor.PopMessageResultFilter.FilterResult; +import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData; +import org.junit.Before; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +public class PopMessageResultFilterImplTest extends BaseActivityTest { + + private static final String CONSUMER_GROUP = "consumerGroup"; + + private ProxyContext ctx; + private SubscriptionData subscriptionData; + + @Before + public void before() throws Throwable { + super.before(); + // BaseActivityTest installs a mock; the default-consumer-settings contract lives on the real instance, + // so replace it with a spy (matching GrpcClientSettingsManagerTest). + grpcClientSettingsManager = spy(new GrpcClientSettingsManager(messagingProcessor)); + ctx = createContext(); + // empty tagsSet subscribes to all tags (SUB_ALL), so the DLQ decision depends only on reconsumeTimes + subscriptionData = mock(SubscriptionData.class); + when(subscriptionData.getTagsSet()).thenReturn(Collections.emptySet()); + } + + private static MessageExt message(int reconsumeTimes) { + MessageExt messageExt = new MessageExt(); + messageExt.setReconsumeTimes(reconsumeTimes); + return messageExt; + } + + @Test + public void testFreshMessageNotRoutedToDlqWithDefaultConsumerMaxAttempts() { + // Regression for apache/rocketmq#8714: when client settings are missing, ReceiveMessageActivity must + // fall back to the real consumer default (maxAttempts = retryMaxTimes + 1), not the protobuf empty + // default whose maxAttempts == 0. With maxAttempts == 0 every message, including a fresh one with + // reconsumeTimes == 0, would be routed straight to the DLQ. + int maxAttempts = grpcClientSettingsManager.getDefaultConsumerSettings().getBackoffPolicy().getMaxAttempts(); + PopMessageResultFilterImpl filter = new PopMessageResultFilterImpl(maxAttempts); + + assertEquals("fresh message (reconsumeTimes == 0) must not be DLQ'd under the real consumer default", + FilterResult.MATCH, filter.filterMessage(ctx, CONSUMER_GROUP, subscriptionData, message(0))); + } + + @Test + public void testZeroMaxAttemptsRoutesFreshMessageToDlq() { + // Documents the bug the PR fixes: the protobuf empty default yields maxAttempts == 0, which DLQs + // a fresh message (reconsumeTimes == 0) because 0 >= 0. Kept as a guard so the fallback never + // regresses back to an empty default. + PopMessageResultFilterImpl filter = new PopMessageResultFilterImpl(0); + assertEquals(FilterResult.TO_DLQ, filter.filterMessage(ctx, CONSUMER_GROUP, subscriptionData, message(0))); + } + + @Test + public void testExhaustedMessageRoutedToDlq() { + // A message that has been retried up to maxAttempts is correctly DLQ'd, confirming the filter still + // enforces the retry ceiling once a real default is in place. + int maxAttempts = grpcClientSettingsManager.getDefaultConsumerSettings().getBackoffPolicy().getMaxAttempts(); + PopMessageResultFilterImpl filter = new PopMessageResultFilterImpl(maxAttempts); + assertEquals(FilterResult.TO_DLQ, + filter.filterMessage(ctx, CONSUMER_GROUP, subscriptionData, message(maxAttempts))); + } +} diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java index 6478f90cb64..69f99cf82f9 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java @@ -49,12 +49,15 @@ import org.apache.rocketmq.proxy.common.MessageReceiptHandle; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.proxy.processor.PopMessageResultFilter; import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest; +import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.service.route.AddressableMessageQueue; import org.apache.rocketmq.proxy.service.route.MessageQueueView; import org.apache.rocketmq.remoting.protocol.route.BrokerData; import org.apache.rocketmq.remoting.protocol.route.QueueData; import org.apache.rocketmq.remoting.protocol.route.TopicRouteData; +import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData; import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentCaptor; @@ -420,6 +423,62 @@ public void testReceiveMessage() { assertEquals(Code.MESSAGE_NOT_FOUND, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues())); } + @Test + public void testReceiveMessageWithMissingClientSettings() { + StreamObserver receiveStreamObserver = mock(ServerCallStreamObserver.class); + ArgumentCaptor responseArgumentCaptor = ArgumentCaptor.forClass(ReceiveMessageResponse.class); + doNothing().when(receiveStreamObserver).onNext(responseArgumentCaptor.capture()); + + when(this.grpcClientSettingsManager.getClientSettings(any())).thenReturn(null); + // ReceiveMessageActivity falls back to the default consumer settings; stub the mock to return the real + // default produced by GrpcClientSettingsManager, mirroring the production code path. + Settings defaultConsumerSettings = new GrpcClientSettingsManager(messagingProcessor).getDefaultConsumerSettings(); + when(this.grpcClientSettingsManager.getDefaultConsumerSettings()).thenReturn(defaultConsumerSettings); + + ArgumentCaptor filterCaptor = ArgumentCaptor.forClass(PopMessageResultFilterImpl.class); + PopResult popResult = new PopResult(PopStatus.NO_NEW_MSG, new ArrayList<>()); + when(this.messagingProcessor.popMessage( + any(), + any(), + anyString(), + anyString(), + anyInt(), + anyLong(), + anyLong(), + anyInt(), + any(), + anyBoolean(), + filterCaptor.capture(), + isNull(), + anyLong())).thenReturn(CompletableFuture.completedFuture(popResult)); + + this.receiveMessageActivity.receiveMessage( + createContext(), + ReceiveMessageRequest.newBuilder() + .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build()) + .setMessageQueue(MessageQueue.newBuilder().setTopic(Resource.newBuilder().setName(TOPIC).build()).build()) + .setAutoRenew(true) + .setFilterExpression(FilterExpression.newBuilder() + .setType(FilterType.TAG) + .setExpression("*") + .build()) + .build(), + receiveStreamObserver + ); + + assertEquals(Code.MESSAGE_NOT_FOUND, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues())); + + // Regression for apache/rocketmq#8714: when client settings are missing, the receive path must fall back + // to the real consumer default, not the protobuf empty default. The filter built for the pop call must + // therefore carry a positive maxAttempts so fresh messages (reconsumeTimes == 0) are not DLQ'd. With the + // old Settings.getDefaultInstance() fallback (maxAttempts == 0) this assertion would fail. + PopMessageResultFilterImpl filter = filterCaptor.getValue(); + MessageExt freshMessage = new MessageExt(); + freshMessage.setReconsumeTimes(0); + assertNotEquals(PopMessageResultFilter.FilterResult.TO_DLQ, + filter.filterMessage(createContext(), CONSUMER_GROUP, new SubscriptionData(), freshMessage)); + } + private Code getResponseCodeFromReceiveMessageResponseList(List responseList) { for (ReceiveMessageResponse response : responseList) { if (response.hasStatus()) {