diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriter.java index 1f63734932..dd2be5f872 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriter.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriter.java @@ -17,9 +17,10 @@ package org.apache.rocketmq.proxy.grpc.v2.common; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; import io.grpc.stub.ServerCallStreamObserver; import io.grpc.stub.StreamObserver; -import java.util.Iterator; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; @@ -27,41 +28,31 @@ import org.apache.rocketmq.logging.InternalLoggerFactory; public class ResponseWriter { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); - public static void write(StreamObserver observer, final Iterator response) { - while (response.hasNext()) { - writeResponse(observer, response.next()); - } - observer.onCompleted(); - } - public static void write(StreamObserver observer, final T response) { - writeResponse(observer, response); - observer.onCompleted(); + if (writeResponse(observer, response)) { + observer.onCompleted(); + } } - public static void writeResponse(StreamObserver observer, final T response) { + public static boolean writeResponse(StreamObserver observer, final T response) { if (null == response) { - return; - } - if (isCancelled(observer)) { - log.warn("client has cancelled the request. response to write: {}", response); - return; + return false; } log.debug("start to write response. response: {}", response); - observer.onNext(response); - } - - public static void writeException(StreamObserver observer, final Throwable e) { - if (null == e) { - return; - } if (isCancelled(observer)) { - log.warn("Client has cancelled the request. Exception to write", e); - return; + log.warn("client has cancelled the request. response to write: {}", response); + return false; } - log.debug("Start to write error response", e); - observer.onError(e); - observer.onCompleted(); + try { + observer.onNext(response); + } catch (StatusRuntimeException statusRuntimeException) { + if (Status.CANCELLED.equals(statusRuntimeException.getStatus())) { + log.warn("client has cancelled the request. response to write: {}", response); + return false; + } + throw statusRuntimeException; + } + return true; } public static boolean isCancelled(StreamObserver observer) { 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 f3965037a8..0bd8bf565a 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 @@ -51,10 +51,7 @@ public class ReceiveMessageActivity extends AbstractMessingActivity { public void receiveMessage(Context ctx, ReceiveMessageRequest request, StreamObserver responseObserver) { ProxyContext proxyContext = createContext(ctx); - ReceiveMessageResponseStreamWriter writer = new ReceiveMessageResponseStreamWriter( - this.messagingProcessor, - responseObserver - ); + ReceiveMessageResponseStreamWriter writer = createWriter(proxyContext, responseObserver); try { Settings settings = this.grpcClientSettingsManager.getClientSettings(proxyContext); @@ -67,7 +64,7 @@ public class ReceiveMessageActivity extends AbstractMessingActivity { pollTime = timeRemaining; } if (pollTime <= 0) { - writer.write(proxyContext, Code.MESSAGE_NOT_FOUND, "time remaining is too small"); + writer.writeAndComplete(proxyContext, Code.MESSAGE_NOT_FOUND, "time remaining is too small"); return; } @@ -85,7 +82,7 @@ public class ReceiveMessageActivity extends AbstractMessingActivity { subscriptionData = FilterAPI.build(topic, filterExpression.getExpression(), GrpcConverter.buildExpressionType(filterExpression.getType())); } catch (Exception e) { - writer.write(proxyContext, Code.ILLEGAL_FILTER_EXPRESSION, e.getMessage()); + writer.writeAndComplete(proxyContext, Code.ILLEGAL_FILTER_EXPRESSION, e.getMessage()); return; } @@ -104,16 +101,23 @@ public class ReceiveMessageActivity extends AbstractMessingActivity { fifo, new PopMessageResultFilterImpl(grpcClientSettingsManager), timeRemaining - ).thenAccept(popResult -> writer.write(proxyContext, request, popResult)) + ).thenAccept(popResult -> writer.writeAndComplete(proxyContext, request, popResult)) .exceptionally(t -> { - writer.write(proxyContext, request, t); + writer.writeAndComplete(proxyContext, request, t); return null; }); } catch (Throwable t) { - writer.write(proxyContext, request, t); + writer.writeAndComplete(proxyContext, request, t); } } + protected ReceiveMessageResponseStreamWriter createWriter(ProxyContext ctx, StreamObserver responseObserver) { + return new ReceiveMessageResponseStreamWriter( + this.messagingProcessor, + responseObserver + ); + } + protected static class ReceiveMessageQueueSelector implements QueueSelector { private final String brokerName; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriter.java index 52514ef0ef..054e9d23e0 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriter.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriter.java @@ -20,14 +20,18 @@ import apache.rocketmq.v2.Code; import apache.rocketmq.v2.Message; import apache.rocketmq.v2.ReceiveMessageRequest; import apache.rocketmq.v2.ReceiveMessageResponse; +import com.google.protobuf.util.Timestamps; import io.grpc.stub.StreamObserver; import java.time.Duration; import java.util.Iterator; import java.util.List; import org.apache.rocketmq.client.consumer.PopResult; import org.apache.rocketmq.client.consumer.PopStatus; +import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.consumer.ReceiptHandle; import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.logging.InternalLogger; +import org.apache.rocketmq.logging.InternalLoggerFactory; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder; @@ -35,7 +39,7 @@ import org.apache.rocketmq.proxy.grpc.v2.common.ResponseWriter; import org.apache.rocketmq.proxy.processor.MessagingProcessor; public class ReceiveMessageResponseStreamWriter { - + private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); protected static final long NACK_INVISIBLE_TIME = Duration.ofSeconds(1).toMillis(); protected final MessagingProcessor messagingProcessor; @@ -48,7 +52,7 @@ public class ReceiveMessageResponseStreamWriter { this.streamObserver = observer; } - public void write(ProxyContext ctx, ReceiveMessageRequest request, PopResult popResult) { + public void writeAndComplete(ProxyContext ctx, ReceiveMessageRequest request, PopResult popResult) { PopStatus status = popResult.getPopStatus(); List messageFoundList = popResult.getMsgFoundList(); try { @@ -62,7 +66,7 @@ public class ReceiveMessageResponseStreamWriter { streamObserver.onNext(ReceiveMessageResponse.newBuilder() .setStatus(ResponseBuilder.buildStatus(Code.OK, Code.OK.name())) .build()); - Iterator messageIterator = messageFoundList.iterator(); + Iterator messageIterator = messageFoundList.iterator(); while (messageIterator.hasNext()) { MessageExt curMessageExt = messageIterator.next(); Message curMessage = convertToMessage(curMessageExt); @@ -93,9 +97,10 @@ public class ReceiveMessageResponseStreamWriter { break; } } catch (Throwable t) { - write(ctx, request, t); + writeResponseWithErrorIgnore( + ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(t)).build()); } finally { - streamObserver.onCompleted(); + onComplete(); } } @@ -121,17 +126,34 @@ public class ReceiveMessageResponseStreamWriter { ); } - public void write(ProxyContext ctx, Code code, String message) { - ResponseWriter.write( - streamObserver, - ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(code, message)).build() - ); + public void writeAndComplete(ProxyContext ctx, Code code, String message) { + writeResponseWithErrorIgnore( + ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(code, message)).build()); + onComplete(); } - public void write(ProxyContext ctx, ReceiveMessageRequest request, Throwable throwable) { - ResponseWriter.write( - streamObserver, - ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(throwable)).build() - ); + public void writeAndComplete(ProxyContext ctx, ReceiveMessageRequest request, Throwable throwable) { + writeResponseWithErrorIgnore( + ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(throwable)).build()); + onComplete(); + } + + protected void writeResponseWithErrorIgnore(ReceiveMessageResponse response) { + try { + ResponseWriter.writeResponse(streamObserver, response); + } catch (Exception e) { + log.error("err when write receive message response", e); + } + } + + protected void onComplete() { + writeResponseWithErrorIgnore(ReceiveMessageResponse.newBuilder() + .setDeliveryTimestamp(Timestamps.fromMillis(System.currentTimeMillis())) + .build()); + try { + streamObserver.onCompleted(); + } catch (Exception e) { + log.error("err when complete receive message response", e); + } } } 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 e9a5457cdd..412e5806c9 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 @@ -94,7 +94,7 @@ public class ReceiveMessageActivityTest extends BaseActivityTest { receiveStreamObserver ); - assertEquals(Code.ILLEGAL_FILTER_EXPRESSION, responseArgumentCaptor.getValue().getStatus().getCode()); + assertEquals(Code.ILLEGAL_FILTER_EXPRESSION, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues())); } @Test @@ -133,7 +133,16 @@ public class ReceiveMessageActivityTest extends BaseActivityTest { .build(), receiveStreamObserver ); - assertEquals(Code.MESSAGE_NOT_FOUND, responseArgumentCaptor.getValue().getStatus().getCode()); + assertEquals(Code.MESSAGE_NOT_FOUND, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues())); + } + + private Code getResponseCodeFromReceiveMessageResponseList(List responseList) { + for (ReceiveMessageResponse response : responseList) { + if (response.hasStatus()) { + return response.getStatus().getCode(); + } + } + return null; } @Test diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriterTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriterTest.java index 1855220a5e..8086b37b36 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriterTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriterTest.java @@ -90,7 +90,7 @@ public class ReceiveMessageResponseStreamWriterTest extends BaseActivityTest { messageExtList.add(createMessageExt(TOPIC, "tag")); messageExtList.add(createMessageExt(TOPIC, "tag")); PopResult popResult = new PopResult(PopStatus.FOUND, messageExtList); - writer.write( + writer.writeAndComplete( ProxyContext.create(), ReceiveMessageRequest.newBuilder() .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build()) @@ -104,7 +104,7 @@ public class ReceiveMessageResponseStreamWriterTest extends BaseActivityTest { ); verify(streamObserver, times(1)).onCompleted(); - verify(streamObserver, times(3)).onNext(any()); + verify(streamObserver, times(4)).onNext(any()); verify(this.messagingProcessor, times(1)) .changeInvisibleTime(any(), any(), anyString(), anyString(), anyString(), anyLong()); @@ -122,7 +122,7 @@ public class ReceiveMessageResponseStreamWriterTest extends BaseActivityTest { doNothing().when(streamObserver).onNext(responseArgumentCaptor.capture()); PopResult popResult = new PopResult(PopStatus.POLLING_FULL, new ArrayList<>()); - writer.write( + writer.writeAndComplete( ProxyContext.create(), ReceiveMessageRequest.newBuilder() .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build()) @@ -135,7 +135,9 @@ public class ReceiveMessageResponseStreamWriterTest extends BaseActivityTest { popResult ); - assertEquals(Code.TOO_MANY_REQUESTS, responseArgumentCaptor.getValue().getStatus().getCode()); + ReceiveMessageResponse response = responseArgumentCaptor.getAllValues().stream().filter(ReceiveMessageResponse::hasStatus) + .findFirst().get(); + assertEquals(Code.TOO_MANY_REQUESTS, response.getStatus().getCode()); } private static MessageExt createMessageExt(String topic, String tags) { diff --git a/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseIT.java b/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseIT.java index 53477c16c4..f1f79bfcdf 100644 --- a/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseIT.java +++ b/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseIT.java @@ -502,14 +502,6 @@ public class GrpcBaseIT extends BaseConf { .build(); } - public AckMessageRequest buildAckMessageRequest(String topic, String group, String messageId, - String receiptHandle) { - return buildAckMessageRequest(topic, group, AckMessageEntry.newBuilder() - .setMessageId(messageId) - .setReceiptHandle(receiptHandle) - .build()); - } - public AckMessageRequest buildAckMessageRequest(String topic, String group, AckMessageEntry... entry) { return AckMessageRequest.newBuilder() .setGroup(Resource.newBuilder() @@ -572,15 +564,6 @@ public class GrpcBaseIT extends BaseConf { return response.get(1).getMessage(); } - public void assertAllAckOk(AckMessageResponse response) { - assertThat(response.getStatus() - .getCode()).isEqualTo(Code.OK); - for (AckMessageResultEntry entry : response.getEntriesList()) { - assertThat(entry.getStatus() - .getCode()).isEqualTo(Code.OK); - } - } - public void assertRecoverOrphanedTransactionCommand(RecoverOrphanedTransactionCommand command, String messageId) { assertThat(command.getOrphanedTransactionalMessage().getSystemProperties().getMessageId()) .isEqualTo(messageId); @@ -606,12 +589,6 @@ public class GrpcBaseIT extends BaseConf { .build(); } - public Settings buildAccessPointClientSettings(int port) { - return Settings.newBuilder() - .setAccessPoint(buildEndpoints(port)) - .build(); - } - public Settings buildSimpleConsumerClientSettings(int maxDeliveryAttempts, boolean fifo) { return Settings.newBuilder() .setClientType(ClientType.SIMPLE_CONSUMER)