Avoid receive failures when client settings are absent

ReceiveMessageActivity can be invoked before cached gRPC client settings are available. Treat missing settings the same as the existing default-settings path so receive validation and pop handling can return a structured response instead of failing with a NullPointerException.

Constraint: Related to #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 test

Not-tested: Full repository test suite

Author: Aias00

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 = Settings.getDefaultInstance();
+            }
             final boolean isLite = ClientType.LITE_PUSH_CONSUMER.equals(settings.getClientType());
 
             Subscription subscription = settings.getSubscription();

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java

@@ -420,6 +420,47 @@ 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);
+
+        PopResult popResult = new PopResult(PopStatus.NO_NEW_MSG, new ArrayList<>());
+        when(this.messagingProcessor.popMessage(
+            any(),
+            any(),
+            anyString(),
+            anyString(),
+            anyInt(),
+            anyLong(),
+            anyLong(),
+            anyInt(),
+            any(),
+            anyBoolean(),
+            any(),
+            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()));
+    }
+
     private Code getResponseCodeFromReceiveMessageResponseList(List<ReceiveMessageResponse> responseList) {
         for (ReceiveMessageResponse response : responseList) {
             if (response.hasStatus()) {