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()) {