Skip to content

Commit 272e93f

Browse files
committed
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
1 parent 8242c1e commit 272e93f

2 files changed

Lines changed: 44 additions & 0 deletions

File tree

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,9 @@ public void receiveMessage(ProxyContext ctx, ReceiveMessageRequest request,
6565

6666
try {
6767
Settings settings = this.grpcClientSettingsManager.getClientSettings(ctx);
68+
if (settings == null) {
69+
settings = Settings.getDefaultInstance();
70+
}
6871
final boolean isLite = ClientType.LITE_PUSH_CONSUMER.equals(settings.getClientType());
6972

7073
Subscription subscription = settings.getSubscription();

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

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -420,6 +420,47 @@ public void testReceiveMessage() {
420420
assertEquals(Code.MESSAGE_NOT_FOUND, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues()));
421421
}
422422

423+
@Test
424+
public void testReceiveMessageWithMissingClientSettings() {
425+
StreamObserver<ReceiveMessageResponse> receiveStreamObserver = mock(ServerCallStreamObserver.class);
426+
ArgumentCaptor<ReceiveMessageResponse> responseArgumentCaptor = ArgumentCaptor.forClass(ReceiveMessageResponse.class);
427+
doNothing().when(receiveStreamObserver).onNext(responseArgumentCaptor.capture());
428+
429+
when(this.grpcClientSettingsManager.getClientSettings(any())).thenReturn(null);
430+
431+
PopResult popResult = new PopResult(PopStatus.NO_NEW_MSG, new ArrayList<>());
432+
when(this.messagingProcessor.popMessage(
433+
any(),
434+
any(),
435+
anyString(),
436+
anyString(),
437+
anyInt(),
438+
anyLong(),
439+
anyLong(),
440+
anyInt(),
441+
any(),
442+
anyBoolean(),
443+
any(),
444+
isNull(),
445+
anyLong())).thenReturn(CompletableFuture.completedFuture(popResult));
446+
447+
this.receiveMessageActivity.receiveMessage(
448+
createContext(),
449+
ReceiveMessageRequest.newBuilder()
450+
.setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build())
451+
.setMessageQueue(MessageQueue.newBuilder().setTopic(Resource.newBuilder().setName(TOPIC).build()).build())
452+
.setAutoRenew(true)
453+
.setFilterExpression(FilterExpression.newBuilder()
454+
.setType(FilterType.TAG)
455+
.setExpression("*")
456+
.build())
457+
.build(),
458+
receiveStreamObserver
459+
);
460+
461+
assertEquals(Code.MESSAGE_NOT_FOUND, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues()));
462+
}
463+
423464
private Code getResponseCodeFromReceiveMessageResponseList(List<ReceiveMessageResponse> responseList) {
424465
for (ReceiveMessageResponse response : responseList) {
425466
if (response.hasStatus()) {

0 commit comments

Comments
 (0)