[ISSUE #8714] Prevent receive message NPE when client settings are missing

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);
     }

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

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);

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)));
+    }
+}

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<ReceiveMessageResponse> receiveStreamObserver = mock(ServerCallStreamObserver.class);
+        ArgumentCaptor<ReceiveMessageResponse> 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<PopMessageResultFilterImpl> 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<ReceiveMessageResponse> responseList) {
         for (ReceiveMessageResponse response : responseList) {
             if (response.hasStatus()) {