From b8b753ba9721be0e7bfc5eefa7fae868c2a99e2e Mon Sep 17 00:00:00 2001 From: zhouxiang Date: Mon, 21 Mar 2022 20:13:10 +0800 Subject: [PATCH] [ISSUE #3949] Add Ack in LocalGrpcTest --- .../common/consumer/ReceiptHandle.java | 4 ++-- .../rocketmq/test/base/GrpcBaseTest.java | 22 +++++++++++++++++-- .../rocketmq/test/proxy/LocalGrpcTest.java | 4 ++++ 3 files changed, 26 insertions(+), 4 deletions(-) diff --git a/common/src/main/java/org/apache/rocketmq/common/consumer/ReceiptHandle.java b/common/src/main/java/org/apache/rocketmq/common/consumer/ReceiptHandle.java index ee7f9a83bf..5b02418f44 100644 --- a/common/src/main/java/org/apache/rocketmq/common/consumer/ReceiptHandle.java +++ b/common/src/main/java/org/apache/rocketmq/common/consumer/ReceiptHandle.java @@ -25,8 +25,8 @@ import org.apache.rocketmq.common.message.MessageExt; public class ReceiptHandle { private static final String SEPARATOR = MessageConst.KEY_SEPARATOR; - private static final String NORMAL_TOPIC = "0"; - private static final String RETRY_TOPIC = "1"; + public static final String NORMAL_TOPIC = "0"; + public static final String RETRY_TOPIC = "1"; private final long startOffset; private final long retrieveTime; private final long invisibleTime; diff --git a/test/src/test/java/org/apache/rocketmq/test/base/GrpcBaseTest.java b/test/src/test/java/org/apache/rocketmq/test/base/GrpcBaseTest.java index bf5db915f5..d46b76ff70 100644 --- a/test/src/test/java/org/apache/rocketmq/test/base/GrpcBaseTest.java +++ b/test/src/test/java/org/apache/rocketmq/test/base/GrpcBaseTest.java @@ -17,8 +17,8 @@ package org.apache.rocketmq.test.base; -import apache.rocketmq.v1.Address; -import apache.rocketmq.v1.AddressScheme; +import apache.rocketmq.v1.AckMessageRequest; +import apache.rocketmq.v1.AckMessageResponse; import apache.rocketmq.v1.Endpoints; import apache.rocketmq.v1.Message; import apache.rocketmq.v1.Partition; @@ -145,6 +145,18 @@ public class GrpcBaseTest extends BaseConf { .build(); } + public AckMessageRequest buildAckMessageRequest(String group, String topic, String receiptHandle) { + return AckMessageRequest.newBuilder() + .setGroup(Resource.newBuilder() + .setName(group) + .build()) + .setTopic(Resource.newBuilder() + .setName(topic) + .build()) + .setReceiptHandle(receiptHandle) + .build(); + } + public void assertQueryRoute(QueryRouteResponse response, int brokerSize) { assertThat(response.getCommon().getStatus().getCode()).isEqualTo(Code.OK_VALUE); assertThat(response.getPartitionsList().size()).isEqualTo(brokerSize * defaultQueueNums); @@ -167,4 +179,10 @@ public class GrpcBaseTest extends BaseConf { .getSystemAttribute() .getMessageId()).isEqualTo(messageId); } + + public void assertAck(AckMessageResponse response) { + assertThat(response.getCommon() + .getStatus() + .getCode()).isEqualTo(Code.OK_VALUE); + } } \ No newline at end of file diff --git a/test/src/test/java/org/apache/rocketmq/test/proxy/LocalGrpcTest.java b/test/src/test/java/org/apache/rocketmq/test/proxy/LocalGrpcTest.java index 16e3d54326..1295f0115c 100644 --- a/test/src/test/java/org/apache/rocketmq/test/proxy/LocalGrpcTest.java +++ b/test/src/test/java/org/apache/rocketmq/test/proxy/LocalGrpcTest.java @@ -17,6 +17,7 @@ package org.apache.rocketmq.test.proxy; +import apache.rocketmq.v1.AckMessageResponse; import apache.rocketmq.v1.MessagingServiceGrpc; import apache.rocketmq.v1.QueryRouteResponse; import apache.rocketmq.v1.ReceiveMessageResponse; @@ -81,5 +82,8 @@ public class LocalGrpcTest extends GrpcBaseTest { ReceiveMessageResponse receiveResponse = blockingStub.withDeadlineAfter(3, TimeUnit.SECONDS) .receiveMessage(buildReceiveMessageRequest(group, broker1Name)); assertReceiveMessage(receiveResponse, messageId); + String receiptHandle = receiveResponse.getMessages(0).getSystemAttribute().getReceiptHandle(); + AckMessageResponse ackMessageResponse = blockingStub.ackMessage(buildAckMessageRequest(group, broker1Name, receiptHandle)); + assertAck(ackMessageResponse); } }