You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by aa...@apache.org on 2022/08/08 08:50:14 UTC
[rocketmq-clients] branch master updated: Extract RPC status checker (#145)
This is an automated email from the ASF dual-hosted git repository.
aaronai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new da6eb7b Extract RPC status checker (#145)
da6eb7b is described below
commit da6eb7ba7ab345f3f989e51400e92bc9607560e1
Author: Aaron Ai <ya...@gmail.com>
AuthorDate: Mon Aug 8 16:50:09 2022 +0800
Extract RPC status checker (#145)
* Extract RPC status checker
* Add more test
---
.../java/exception/PaymentRequiredException.java | 26 +
.../StatusChecker.java} | 84 +--
.../rocketmq/client/java/impl/ClientImpl.java | 40 +-
.../client/java/impl/consumer/ConsumerImpl.java | 9 +-
.../java/impl/consumer/PushConsumerImpl.java | 40 +-
.../java/impl/consumer/ReceiveMessageResult.java | 46 +-
.../java/impl/consumer/SimpleConsumerImpl.java | 73 +--
.../client/java/impl/producer/SendReceiptImpl.java | 71 +--
.../client/java/exception/StatusCheckerTest.java | 564 +++++++++++++++++++++
.../java/impl/consumer/ProcessQueueImplTest.java | 11 +-
10 files changed, 678 insertions(+), 286 deletions(-)
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/exception/PaymentRequiredException.java b/java/client/src/main/java/org/apache/rocketmq/client/java/exception/PaymentRequiredException.java
new file mode 100644
index 0000000..f5802ce
--- /dev/null
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/exception/PaymentRequiredException.java
@@ -0,0 +1,26 @@
+/*
+ * 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.client.java.exception;
+
+import org.apache.rocketmq.client.apis.ClientException;
+
+public class PaymentRequiredException extends ClientException {
+ public PaymentRequiredException(int responseCode, String requestId, String message) {
+ super(responseCode, requestId, message);
+ }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ReceiveMessageResult.java b/java/client/src/main/java/org/apache/rocketmq/client/java/exception/StatusChecker.java
similarity index 50%
copy from java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ReceiveMessageResult.java
copy to java/client/src/main/java/org/apache/rocketmq/client/java/exception/StatusChecker.java
index cad65b2..e22397b 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ReceiveMessageResult.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/exception/StatusChecker.java
@@ -15,76 +15,90 @@
* limitations under the License.
*/
-package org.apache.rocketmq.client.java.impl.consumer;
+package org.apache.rocketmq.client.java.exception;
import apache.rocketmq.v2.Code;
+import apache.rocketmq.v2.ReceiveMessageResponse;
import apache.rocketmq.v2.Status;
-import java.util.ArrayList;
-import java.util.List;
import org.apache.rocketmq.client.apis.ClientException;
-import org.apache.rocketmq.client.apis.message.MessageView;
-import org.apache.rocketmq.client.java.exception.BadRequestException;
-import org.apache.rocketmq.client.java.exception.ForbiddenException;
-import org.apache.rocketmq.client.java.exception.InternalErrorException;
-import org.apache.rocketmq.client.java.exception.NotFoundException;
-import org.apache.rocketmq.client.java.exception.ProxyTimeoutException;
-import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
-import org.apache.rocketmq.client.java.exception.UnauthorizedException;
-import org.apache.rocketmq.client.java.exception.UnsupportedException;
-import org.apache.rocketmq.client.java.message.MessageViewImpl;
-import org.apache.rocketmq.client.java.route.Endpoints;
+import org.apache.rocketmq.client.java.misc.MetadataUtils;
+import org.apache.rocketmq.client.java.rpc.RpcInvocation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-public class ReceiveMessageResult {
- private final Endpoints endpoints;
- private final List<MessageViewImpl> messages;
+public class StatusChecker {
+ private static final Logger LOGGER = LoggerFactory.getLogger(StatusChecker.class);
- public ReceiveMessageResult(Endpoints endpoints, String requestId, Status status, List<MessageViewImpl> messages)
- throws ClientException {
- this.endpoints = endpoints;
+ private StatusChecker() {
+ }
+
+ public static void check(Status status, RpcInvocation<?> invocation) throws ClientException {
+ final String requestId = invocation.getContext().getRequestId();
final Code code = status.getCode();
final int codeNumber = code.getNumber();
final String statusMessage = status.getMessage();
switch (code) {
case OK:
- case MESSAGE_NOT_FOUND:
- break;
+ case MULTIPLE_RESULTS:
+ return;
case BAD_REQUEST:
+ case ILLEGAL_ACCESS_POINT:
case ILLEGAL_TOPIC:
case ILLEGAL_CONSUMER_GROUP:
+ case ILLEGAL_MESSAGE_TAG:
+ case ILLEGAL_MESSAGE_KEY:
+ case ILLEGAL_MESSAGE_GROUP:
+ case ILLEGAL_MESSAGE_PROPERTY_KEY:
+ case INVALID_TRANSACTION_ID:
+ case ILLEGAL_MESSAGE_ID:
case ILLEGAL_FILTER_EXPRESSION:
case ILLEGAL_INVISIBLE_TIME:
+ case ILLEGAL_DELIVERY_TIME:
+ case INVALID_RECEIPT_HANDLE:
+ case MESSAGE_PROPERTY_CONFLICT_WITH_TYPE:
+ case UNRECOGNIZED_CLIENT_TYPE:
+ case MESSAGE_CORRUPTED:
case CLIENT_ID_REQUIRED:
throw new BadRequestException(codeNumber, requestId, statusMessage);
case UNAUTHORIZED:
throw new UnauthorizedException(codeNumber, requestId, statusMessage);
+ case PAYMENT_REQUIRED:
+ throw new PaymentRequiredException(codeNumber, requestId, statusMessage);
case FORBIDDEN:
throw new ForbiddenException(codeNumber, requestId, statusMessage);
+ case MESSAGE_NOT_FOUND:
+ if (invocation.getResponse() instanceof ReceiveMessageResponse) {
+ return;
+ }
+ // fall through on purpose.
case NOT_FOUND:
case TOPIC_NOT_FOUND:
case CONSUMER_GROUP_NOT_FOUND:
throw new NotFoundException(codeNumber, requestId, statusMessage);
+ case PAYLOAD_TOO_LARGE:
+ case MESSAGE_BODY_TOO_LARGE:
+ throw new PayloadTooLargeException(codeNumber, requestId, statusMessage);
case TOO_MANY_REQUESTS:
throw new TooManyRequestsException(codeNumber, requestId, statusMessage);
+ case REQUEST_HEADER_FIELDS_TOO_LARGE:
+ case MESSAGE_PROPERTIES_TOO_LARGE:
+ throw new RequestHeaderFieldsTooLargeException(codeNumber, requestId, statusMessage);
case INTERNAL_ERROR:
case INTERNAL_SERVER_ERROR:
+ case HA_NOT_AVAILABLE:
throw new InternalErrorException(codeNumber, requestId, statusMessage);
case PROXY_TIMEOUT:
+ case MASTER_PERSISTENCE_TIMEOUT:
+ case SLAVE_PERSISTENCE_TIMEOUT:
throw new ProxyTimeoutException(codeNumber, requestId, statusMessage);
+ case UNSUPPORTED:
+ case VERSION_UNSUPPORTED:
+ case VERIFY_FIFO_MESSAGE_UNSUPPORTED:
+ throw new UnsupportedException(codeNumber, requestId, statusMessage);
default:
+ LOGGER.warn("Unrecognized status code={}, requestId={}, statusMessage={}, clientVersion={}",
+ codeNumber, requestId, statusMessage, MetadataUtils.getVersion());
throw new UnsupportedException(codeNumber, requestId, statusMessage);
}
- this.messages = messages;
- }
-
- public List<MessageView> getMessageViews() {
- return new ArrayList<>(messages);
- }
-
- public Endpoints getEndpoints() {
- return endpoints;
- }
-
- public List<MessageViewImpl> getMessageViewImpls() {
- return messages;
}
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
index da1ff6f..d67ae57 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
@@ -71,12 +71,8 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
-import org.apache.rocketmq.client.java.exception.BadRequestException;
import org.apache.rocketmq.client.java.exception.InternalErrorException;
-import org.apache.rocketmq.client.java.exception.NotFoundException;
-import org.apache.rocketmq.client.java.exception.ProxyTimeoutException;
-import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
-import org.apache.rocketmq.client.java.exception.UnsupportedException;
+import org.apache.rocketmq.client.java.exception.StatusChecker;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
import org.apache.rocketmq.client.java.hook.MessageInterceptor;
@@ -178,6 +174,7 @@ public abstract class ClientImpl extends AbstractIdleService implements Client,
}));
}
+
/**
* Start the rocketmq client and do some preparatory work.
*/
@@ -188,9 +185,8 @@ public abstract class ClientImpl extends AbstractIdleService implements Client,
// Fetch topic route from remote.
LOGGER.info("Begin to fetch topic(s) route data from remote during client startup, clientId={}, topics={}",
clientId, topics);
- final List<ListenableFuture<TopicRouteData>> futures =
- topics.stream().map(this::fetchTopicRoute).collect(Collectors.toList());
- for (ListenableFuture<TopicRouteData> future : futures) {
+ for (String topic : topics) {
+ final ListenableFuture<TopicRouteData> future = fetchTopicRoute(topic);
future.get();
}
LOGGER.info("Fetch topic route data from remote successfully during startup, clientId={}, topics={}",
@@ -625,7 +621,7 @@ public abstract class ClientImpl extends AbstractIdleService implements Client,
return future;
}
- private ListenableFuture<TopicRouteData> fetchTopicRoute0(final String topic) {
+ ListenableFuture<TopicRouteData> fetchTopicRoute0(final String topic) {
try {
Resource topicResource = Resource.newBuilder().setName(topic).build();
final QueryRouteRequest request = QueryRouteRequest.newBuilder().setTopic(topicResource)
@@ -635,32 +631,8 @@ public abstract class ClientImpl extends AbstractIdleService implements Client,
clientManager.queryRoute(endpoints, metadata, request, clientConfiguration.getRequestTimeout());
return Futures.transformAsync(future, invocation -> {
final QueryRouteResponse response = invocation.getResponse();
- final String requestId = invocation.getContext().getRequestId();
final Status status = response.getStatus();
- final String statusMessage = status.getMessage();
- final Code code = status.getCode();
- final int codeNumber = code.getNumber();
- switch (code) {
- case OK:
- break;
- case BAD_REQUEST:
- case ILLEGAL_ACCESS_POINT:
- case ILLEGAL_TOPIC:
- case CLIENT_ID_REQUIRED:
- throw new BadRequestException(codeNumber, requestId, statusMessage);
- case NOT_FOUND:
- case TOPIC_NOT_FOUND:
- throw new NotFoundException(codeNumber, requestId, statusMessage);
- case TOO_MANY_REQUESTS:
- throw new TooManyRequestsException(codeNumber, requestId, statusMessage);
- case INTERNAL_ERROR:
- case INTERNAL_SERVER_ERROR:
- throw new InternalErrorException(codeNumber, requestId, statusMessage);
- case PROXY_TIMEOUT:
- throw new ProxyTimeoutException(codeNumber, requestId, statusMessage);
- default:
- throw new UnsupportedException(codeNumber, requestId, statusMessage);
- }
+ StatusChecker.check(status, invocation);
final List<MessageQueue> messageQueuesList = response.getMessageQueuesList();
final TopicRouteData topicRouteData = new TopicRouteData(messageQueuesList);
return Futures.immediateFuture(topicRouteData);
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumerImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumerImpl.java
index fbc6191..791a8f2 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumerImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumerImpl.java
@@ -50,6 +50,7 @@ import java.util.regex.Pattern;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.message.MessageId;
+import org.apache.rocketmq.client.java.exception.StatusChecker;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
import org.apache.rocketmq.client.java.impl.ClientImpl;
@@ -83,8 +84,8 @@ abstract class ConsumerImpl extends ClientImpl {
final Duration timeout = Duration.ofNanos(awaitDuration.toNanos() + tolerance.toNanos());
final ListenableFuture<RpcInvocation<Iterator<ReceiveMessageResponse>>> future =
clientManager.receiveMessage(endpoints, metadata, request, timeout);
- return Futures.transformAsync(future, context -> {
- final Iterator<ReceiveMessageResponse> it = context.getResponse();
+ return Futures.transformAsync(future, invocation -> {
+ final Iterator<ReceiveMessageResponse> it = invocation.getResponse();
Status status = Status.newBuilder().setCode(Code.INTERNAL_SERVER_ERROR)
.setMessage("status was not set by server")
.build();
@@ -111,8 +112,8 @@ abstract class ConsumerImpl extends ClientImpl {
final MessageViewImpl view = MessageViewImpl.fromProtobuf(message, mq, deliveryTimestampFromRemote);
messages.add(view);
}
- final ReceiveMessageResult receiveMessageResult = new ReceiveMessageResult(endpoints,
- context.getContext().getRequestId(), status, messages);
+ StatusChecker.check(status, invocation);
+ final ReceiveMessageResult receiveMessageResult = new ReceiveMessageResult(endpoints, messages);
return Futures.immediateFuture(receiveMessageResult);
}, MoreExecutors.directExecutor());
} catch (Throwable t) {
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
index 3d17ea9..d4f329c 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
@@ -58,13 +58,7 @@ import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.client.apis.message.MessageId;
-import org.apache.rocketmq.client.java.exception.BadRequestException;
-import org.apache.rocketmq.client.java.exception.ForbiddenException;
-import org.apache.rocketmq.client.java.exception.InternalErrorException;
-import org.apache.rocketmq.client.java.exception.NotFoundException;
-import org.apache.rocketmq.client.java.exception.ProxyTimeoutException;
-import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
-import org.apache.rocketmq.client.java.exception.UnsupportedException;
+import org.apache.rocketmq.client.java.exception.StatusChecker;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
import org.apache.rocketmq.client.java.impl.ClientSettings;
@@ -278,36 +272,10 @@ class PushConsumerImpl extends ConsumerImpl implements PushConsumer, MessageCach
final Duration requestTimeout = clientConfiguration.getRequestTimeout();
return clientManager.queryAssignment(endpoints, metadata, request, requestTimeout);
}, MoreExecutors.directExecutor());
- return Futures.transformAsync(responseFuture, context -> {
- final QueryAssignmentResponse response = context.getResponse();
+ return Futures.transformAsync(responseFuture, invocation -> {
+ final QueryAssignmentResponse response = invocation.getResponse();
final Status status = response.getStatus();
- final Code code = status.getCode();
- final int codeNumber = code.getNumber();
- final String requestId = context.getContext().getRequestId();
- final String statusMessage = status.getMessage();
- switch (code) {
- case OK:
- break;
- case BAD_REQUEST:
- case ILLEGAL_ACCESS_POINT:
- case ILLEGAL_TOPIC:
- case CLIENT_ID_REQUIRED:
- throw new BadRequestException(codeNumber, requestId, statusMessage);
- case FORBIDDEN:
- throw new ForbiddenException(codeNumber, requestId, statusMessage);
- case NOT_FOUND:
- case TOPIC_NOT_FOUND:
- throw new NotFoundException(codeNumber, requestId, statusMessage);
- case TOO_MANY_REQUESTS:
- throw new TooManyRequestsException(codeNumber, requestId, statusMessage);
- case INTERNAL_ERROR:
- case INTERNAL_SERVER_ERROR:
- throw new InternalErrorException(codeNumber, requestId, statusMessage);
- case PROXY_TIMEOUT:
- throw new ProxyTimeoutException(codeNumber, requestId, statusMessage);
- default:
- throw new UnsupportedException(codeNumber, requestId, statusMessage);
- }
+ StatusChecker.check(status, invocation);
final List<Assignment> assignmentList = response.getAssignmentsList().stream().map(assignment ->
new Assignment(new MessageQueueImpl(assignment.getMessageQueue()))).collect(Collectors.toList());
final Assignments assignments = new Assignments(assignmentList);
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ReceiveMessageResult.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ReceiveMessageResult.java
index cad65b2..ab11ff9 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ReceiveMessageResult.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ReceiveMessageResult.java
@@ -17,20 +17,9 @@
package org.apache.rocketmq.client.java.impl.consumer;
-import apache.rocketmq.v2.Code;
-import apache.rocketmq.v2.Status;
import java.util.ArrayList;
import java.util.List;
-import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.message.MessageView;
-import org.apache.rocketmq.client.java.exception.BadRequestException;
-import org.apache.rocketmq.client.java.exception.ForbiddenException;
-import org.apache.rocketmq.client.java.exception.InternalErrorException;
-import org.apache.rocketmq.client.java.exception.NotFoundException;
-import org.apache.rocketmq.client.java.exception.ProxyTimeoutException;
-import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
-import org.apache.rocketmq.client.java.exception.UnauthorizedException;
-import org.apache.rocketmq.client.java.exception.UnsupportedException;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.route.Endpoints;
@@ -38,41 +27,8 @@ public class ReceiveMessageResult {
private final Endpoints endpoints;
private final List<MessageViewImpl> messages;
- public ReceiveMessageResult(Endpoints endpoints, String requestId, Status status, List<MessageViewImpl> messages)
- throws ClientException {
+ public ReceiveMessageResult(Endpoints endpoints, List<MessageViewImpl> messages) {
this.endpoints = endpoints;
- final Code code = status.getCode();
- final int codeNumber = code.getNumber();
- final String statusMessage = status.getMessage();
- switch (code) {
- case OK:
- case MESSAGE_NOT_FOUND:
- break;
- case BAD_REQUEST:
- case ILLEGAL_TOPIC:
- case ILLEGAL_CONSUMER_GROUP:
- case ILLEGAL_FILTER_EXPRESSION:
- case ILLEGAL_INVISIBLE_TIME:
- case CLIENT_ID_REQUIRED:
- throw new BadRequestException(codeNumber, requestId, statusMessage);
- case UNAUTHORIZED:
- throw new UnauthorizedException(codeNumber, requestId, statusMessage);
- case FORBIDDEN:
- throw new ForbiddenException(codeNumber, requestId, statusMessage);
- case NOT_FOUND:
- case TOPIC_NOT_FOUND:
- case CONSUMER_GROUP_NOT_FOUND:
- throw new NotFoundException(codeNumber, requestId, statusMessage);
- case TOO_MANY_REQUESTS:
- throw new TooManyRequestsException(codeNumber, requestId, statusMessage);
- case INTERNAL_ERROR:
- case INTERNAL_SERVER_ERROR:
- throw new InternalErrorException(codeNumber, requestId, statusMessage);
- case PROXY_TIMEOUT:
- throw new ProxyTimeoutException(codeNumber, requestId, statusMessage);
- default:
- throw new UnsupportedException(codeNumber, requestId, statusMessage);
- }
this.messages = messages;
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/SimpleConsumerImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/SimpleConsumerImpl.java
index 9fad3a1..e5f9618 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/SimpleConsumerImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/SimpleConsumerImpl.java
@@ -19,7 +19,6 @@ package org.apache.rocketmq.client.java.impl.consumer;
import apache.rocketmq.v2.AckMessageResponse;
import apache.rocketmq.v2.ChangeInvisibleDurationResponse;
-import apache.rocketmq.v2.Code;
import apache.rocketmq.v2.ReceiveMessageRequest;
import apache.rocketmq.v2.Status;
import com.google.common.math.IntMath;
@@ -43,14 +42,7 @@ import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageView;
-import org.apache.rocketmq.client.java.exception.BadRequestException;
-import org.apache.rocketmq.client.java.exception.ForbiddenException;
-import org.apache.rocketmq.client.java.exception.InternalErrorException;
-import org.apache.rocketmq.client.java.exception.NotFoundException;
-import org.apache.rocketmq.client.java.exception.ProxyTimeoutException;
-import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
-import org.apache.rocketmq.client.java.exception.UnauthorizedException;
-import org.apache.rocketmq.client.java.exception.UnsupportedException;
+import org.apache.rocketmq.client.java.exception.StatusChecker;
import org.apache.rocketmq.client.java.impl.ClientSettings;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.message.protocol.Resource;
@@ -242,38 +234,10 @@ class SimpleConsumerImpl extends ConsumerImpl implements SimpleConsumer {
MessageViewImpl impl = (MessageViewImpl) messageView;
final ListenableFuture<RpcInvocation<AckMessageResponse>> future = ackMessage(impl);
return Futures.transformAsync(future, invocation -> {
- final String requestId = invocation.getContext().getRequestId();
final AckMessageResponse response = invocation.getResponse();
final Status status = response.getStatus();
- final Code code = status.getCode();
- final int codeNumber = code.getNumber();
- final String statusMessage = status.getMessage();
- switch (code) {
- case OK:
- return Futures.immediateVoidFuture();
- case BAD_REQUEST:
- case ILLEGAL_TOPIC:
- case ILLEGAL_CONSUMER_GROUP:
- case INVALID_RECEIPT_HANDLE:
- case CLIENT_ID_REQUIRED:
- throw new BadRequestException(codeNumber, requestId, statusMessage);
- case UNAUTHORIZED:
- throw new UnauthorizedException(codeNumber, requestId, statusMessage);
- case FORBIDDEN:
- throw new ForbiddenException(codeNumber, requestId, statusMessage);
- case NOT_FOUND:
- case TOPIC_NOT_FOUND:
- throw new NotFoundException(codeNumber, requestId, statusMessage);
- case TOO_MANY_REQUESTS:
- throw new TooManyRequestsException(codeNumber, requestId, statusMessage);
- case INTERNAL_ERROR:
- case INTERNAL_SERVER_ERROR:
- throw new InternalErrorException(codeNumber, requestId, statusMessage);
- case PROXY_TIMEOUT:
- throw new ProxyTimeoutException(codeNumber, requestId, statusMessage);
- default:
- throw new UnsupportedException(codeNumber, requestId, statusMessage);
- }
+ StatusChecker.check(status, invocation);
+ return Futures.immediateVoidFuture();
}, clientCallbackExecutor);
}
@@ -313,38 +277,11 @@ class SimpleConsumerImpl extends ConsumerImpl implements SimpleConsumer {
changeInvisibleDuration(impl, invisibleDuration);
return Futures.transformAsync(future, invocation -> {
final ChangeInvisibleDurationResponse response = invocation.getResponse();
- final String requestId = invocation.getContext().getRequestId();
// Refresh receipt handle manually.
impl.setReceiptHandle(response.getReceiptHandle());
final Status status = response.getStatus();
- final Code code = status.getCode();
- final int codeNumber = code.getNumber();
- final String statusMessage = status.getMessage();
- switch (code) {
- case OK:
- return Futures.immediateVoidFuture();
- case BAD_REQUEST:
- case ILLEGAL_TOPIC:
- case ILLEGAL_CONSUMER_GROUP:
- case ILLEGAL_INVISIBLE_TIME:
- case INVALID_RECEIPT_HANDLE:
- case CLIENT_ID_REQUIRED:
- throw new BadRequestException(codeNumber, requestId, statusMessage);
- case UNAUTHORIZED:
- throw new UnauthorizedException(codeNumber, requestId, statusMessage);
- case NOT_FOUND:
- case TOPIC_NOT_FOUND:
- throw new NotFoundException(codeNumber, requestId, statusMessage);
- case TOO_MANY_REQUESTS:
- throw new TooManyRequestsException(codeNumber, requestId, statusMessage);
- case INTERNAL_ERROR:
- case INTERNAL_SERVER_ERROR:
- throw new InternalErrorException(codeNumber, requestId, statusMessage);
- case PROXY_TIMEOUT:
- throw new ProxyTimeoutException(codeNumber, requestId, statusMessage);
- default:
- throw new UnsupportedException(codeNumber, requestId, statusMessage);
- }
+ StatusChecker.check(status, invocation);
+ return Futures.immediateVoidFuture();
}, MoreExecutors.directExecutor());
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/SendReceiptImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/SendReceiptImpl.java
index 33e4856..2ed3cc7 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/SendReceiptImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/SendReceiptImpl.java
@@ -29,16 +29,7 @@ import java.util.Optional;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
-import org.apache.rocketmq.client.java.exception.BadRequestException;
-import org.apache.rocketmq.client.java.exception.ForbiddenException;
-import org.apache.rocketmq.client.java.exception.InternalErrorException;
-import org.apache.rocketmq.client.java.exception.NotFoundException;
-import org.apache.rocketmq.client.java.exception.PayloadTooLargeException;
-import org.apache.rocketmq.client.java.exception.ProxyTimeoutException;
-import org.apache.rocketmq.client.java.exception.RequestHeaderFieldsTooLargeException;
-import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
-import org.apache.rocketmq.client.java.exception.UnauthorizedException;
-import org.apache.rocketmq.client.java.exception.UnsupportedException;
+import org.apache.rocketmq.client.java.exception.StatusChecker;
import org.apache.rocketmq.client.java.message.MessageIdCodec;
import org.apache.rocketmq.client.java.route.Endpoints;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
@@ -81,7 +72,6 @@ public class SendReceiptImpl implements SendReceipt {
public static List<SendReceiptImpl> processSendMessageResponseInvocation(MessageQueueImpl mq,
RpcInvocation<SendMessageResponse> invocation) throws ClientException {
- final String requestId = invocation.getContext().getRequestId();
final SendMessageResponse response = invocation.getResponse();
Status status = response.getStatus();
List<SendReceiptImpl> sendReceipts = new ArrayList<>();
@@ -92,58 +82,15 @@ public class SendReceiptImpl implements SendReceipt {
if (abnormalStatus.isPresent()) {
status = abnormalStatus.get();
}
- final Code code = status.getCode();
- if (code.equals(Code.OK)) {
- for (SendResultEntry entry : entries) {
- final MessageId messageId = MessageIdCodec.getInstance().decode(entry.getMessageId());
- final String transactionId = entry.getTransactionId();
- final long offset = entry.getOffset();
- final SendReceiptImpl impl = new SendReceiptImpl(messageId, transactionId, mq, offset);
- sendReceipts.add(impl);
- }
- return sendReceipts;
- }
- final int codeNumber = code.getNumber();
- final String statusMessage = status.getMessage();
- switch (code) {
- case ILLEGAL_TOPIC:
- case ILLEGAL_MESSAGE_TAG:
- case ILLEGAL_MESSAGE_KEY:
- case ILLEGAL_MESSAGE_GROUP:
- case ILLEGAL_MESSAGE_PROPERTY_KEY:
- case ILLEGAL_MESSAGE_ID:
- case ILLEGAL_DELIVERY_TIME:
- case MESSAGE_PROPERTY_CONFLICT_WITH_TYPE:
- case MESSAGE_CORRUPTED:
- case CLIENT_ID_REQUIRED:
- throw new BadRequestException(codeNumber, requestId, statusMessage);
- case UNAUTHORIZED:
- throw new UnauthorizedException(codeNumber, requestId, statusMessage);
- case FORBIDDEN:
- throw new ForbiddenException(codeNumber, requestId, statusMessage);
- case NOT_FOUND:
- case TOPIC_NOT_FOUND:
- throw new NotFoundException(codeNumber, requestId, statusMessage);
- case PAYLOAD_TOO_LARGE:
- case MESSAGE_BODY_TOO_LARGE:
- throw new PayloadTooLargeException(codeNumber, requestId, statusMessage);
- case TOO_MANY_REQUESTS:
- throw new TooManyRequestsException(codeNumber, requestId, statusMessage);
- case REQUEST_HEADER_FIELDS_TOO_LARGE:
- case MESSAGE_PROPERTIES_TOO_LARGE:
- throw new RequestHeaderFieldsTooLargeException(codeNumber, requestId, statusMessage);
- case INTERNAL_ERROR:
- case INTERNAL_SERVER_ERROR:
- case HA_NOT_AVAILABLE:
- throw new InternalErrorException(codeNumber, requestId, statusMessage);
- case PROXY_TIMEOUT:
- case MASTER_PERSISTENCE_TIMEOUT:
- case SLAVE_PERSISTENCE_TIMEOUT:
- throw new ProxyTimeoutException(codeNumber, requestId, statusMessage);
- case UNSUPPORTED:
- default:
- throw new UnsupportedException(codeNumber, requestId, statusMessage);
+ StatusChecker.check(status, invocation);
+ for (SendResultEntry entry : entries) {
+ final MessageId messageId = MessageIdCodec.getInstance().decode(entry.getMessageId());
+ final String transactionId = entry.getTransactionId();
+ final long offset = entry.getOffset();
+ final SendReceiptImpl impl = new SendReceiptImpl(messageId, transactionId, mq, offset);
+ sendReceipts.add(impl);
}
+ return sendReceipts;
}
@Override
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/exception/StatusCheckerTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/exception/StatusCheckerTest.java
new file mode 100644
index 0000000..de37e60
--- /dev/null
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/exception/StatusCheckerTest.java
@@ -0,0 +1,564 @@
+/*
+ * 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.client.java.exception;
+
+import static org.junit.Assert.fail;
+
+import apache.rocketmq.v2.Code;
+import apache.rocketmq.v2.ReceiveMessageResponse;
+import apache.rocketmq.v2.Status;
+import io.grpc.Metadata;
+import org.apache.rocketmq.client.apis.ClientException;
+import org.apache.rocketmq.client.java.misc.RequestIdGenerator;
+import org.apache.rocketmq.client.java.route.Endpoints;
+import org.apache.rocketmq.client.java.rpc.Context;
+import org.apache.rocketmq.client.java.rpc.RpcInvocation;
+import org.apache.rocketmq.client.java.rpc.Signature;
+import org.apache.rocketmq.client.java.tool.TestBase;
+import org.junit.Test;
+
+public class StatusCheckerTest extends TestBase {
+ private Context generateContext() {
+ final Endpoints endpoints = fakeEndpoints();
+ Metadata metadata = new Metadata();
+ metadata.put(Metadata.Key.of(Signature.REQUEST_ID_KEY, Metadata.ASCII_STRING_MARSHALLER),
+ RequestIdGenerator.getInstance().next());
+ return new Context(endpoints, metadata);
+ }
+
+ @Test
+ public void testOK() throws ClientException {
+ Status status = Status.newBuilder().setCode(Code.OK).build();
+ Object response = new Object();
+ final Context context = generateContext();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ StatusChecker.check(status, invocation);
+ }
+
+ @Test
+ public void testMultipleResults() throws ClientException {
+ Status status = Status.newBuilder().setCode(Code.OK).build();
+ Object response = new Object();
+ final Context context = generateContext();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ StatusChecker.check(status, invocation);
+ }
+
+ @Test
+ public void testBadRequest() throws ClientException {
+ Object response = new Object();
+ final Context context = generateContext();
+
+ {
+ Status status = Status.newBuilder().setCode(Code.BAD_REQUEST).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_ACCESS_POINT).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_TOPIC).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_CONSUMER_GROUP).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_MESSAGE_TAG).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_MESSAGE_KEY).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_MESSAGE_GROUP).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_MESSAGE_PROPERTY_KEY).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.INVALID_TRANSACTION_ID).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_MESSAGE_ID).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_MESSAGE_GROUP).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_FILTER_EXPRESSION).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_INVISIBLE_TIME).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.ILLEGAL_DELIVERY_TIME).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.INVALID_RECEIPT_HANDLE).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.UNRECOGNIZED_CLIENT_TYPE).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.MESSAGE_CORRUPTED).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.CLIENT_ID_REQUIRED).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (BadRequestException ignore) {
+ // ignore on purpose
+ }
+ }
+ }
+
+ @Test
+ public void testUnauthorized() throws ClientException {
+ Status status = Status.newBuilder().setCode(Code.UNAUTHORIZED).build();
+ Object response = new Object();
+ final Context context = generateContext();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (UnauthorizedException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ @Test
+ public void testPaymentRequired() throws ClientException {
+ Status status = Status.newBuilder().setCode(Code.PAYMENT_REQUIRED).build();
+ Object response = new Object();
+ final Context context = generateContext();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (PaymentRequiredException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ @Test
+ public void testForbidden() throws ClientException {
+ Status status = Status.newBuilder().setCode(Code.FORBIDDEN).build();
+ Object response = new Object();
+ final Context context = generateContext();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ForbiddenException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ @Test
+ public void testMessageNotFoundDuringReceiving() throws ClientException {
+ Status status = Status.newBuilder().setCode(Code.MESSAGE_NOT_FOUND).build();
+ ReceiveMessageResponse response = ReceiveMessageResponse.newBuilder().build();
+ final Context context = generateContext();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ StatusChecker.check(status, invocation);
+ }
+
+ @Test
+ public void testNotFound() {
+ Object response = new Object();
+ final Context context = generateContext();
+
+ {
+ Status status = Status.newBuilder().setCode(Code.MESSAGE_NOT_FOUND).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ClientException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.NOT_FOUND).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ClientException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.TOPIC_NOT_FOUND).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ClientException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.CONSUMER_GROUP_NOT_FOUND).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ClientException ignore) {
+ // ignore on purpose
+ }
+ }
+ }
+
+ @Test
+ public void testPayloadTooLarge() {
+ Object response = new Object();
+ final Context context = generateContext();
+
+ {
+ Status status = Status.newBuilder().setCode(Code.PAYLOAD_TOO_LARGE).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ClientException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.MESSAGE_BODY_TOO_LARGE).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ClientException ignore) {
+ // ignore on purpose
+ }
+ }
+ }
+
+ @Test
+ public void testTooManyRequests() throws ClientException {
+ Status status = Status.newBuilder().setCode(Code.TOO_MANY_REQUESTS).build();
+ ReceiveMessageResponse response = ReceiveMessageResponse.newBuilder().build();
+ final Context context = generateContext();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (TooManyRequestsException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ @Test
+ public void testRequestHeaderFieldsTooLarge() throws ClientException {
+ Object response = new Object();
+ final Context context = generateContext();
+
+ {
+ Status status = Status.newBuilder().setCode(Code.REQUEST_HEADER_FIELDS_TOO_LARGE).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (RequestHeaderFieldsTooLargeException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.MESSAGE_PROPERTIES_TOO_LARGE).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (RequestHeaderFieldsTooLargeException ignore) {
+ // ignore on purpose
+ }
+ }
+ }
+
+ @Test
+ public void testInternalError() throws ClientException {
+ Object response = new Object();
+ final Context context = generateContext();
+
+ {
+ Status status = Status.newBuilder().setCode(Code.INTERNAL_ERROR).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (InternalErrorException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.INTERNAL_SERVER_ERROR).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (InternalErrorException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.HA_NOT_AVAILABLE).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (InternalErrorException ignore) {
+ // ignore on purpose
+ }
+ }
+ }
+
+ @Test
+ public void testProxyTimeout() throws ClientException {
+ Object response = new Object();
+ final Context context = generateContext();
+
+ {
+ Status status = Status.newBuilder().setCode(Code.PROXY_TIMEOUT).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ProxyTimeoutException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.MASTER_PERSISTENCE_TIMEOUT).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ProxyTimeoutException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.SLAVE_PERSISTENCE_TIMEOUT).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (ProxyTimeoutException ignore) {
+ // ignore on purpose
+ }
+ }
+ }
+
+ @Test
+ public void testUnsupported() throws ClientException {
+ Object response = new Object();
+ final Context context = generateContext();
+
+ {
+ Status status = Status.newBuilder().setCode(Code.UNSUPPORTED).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (UnsupportedException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.VERSION_UNSUPPORTED).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (UnsupportedException ignore) {
+ // ignore on purpose
+ }
+ }
+
+ {
+ Status status = Status.newBuilder().setCode(Code.VERIFY_FIFO_MESSAGE_UNSUPPORTED).build();
+ RpcInvocation<Object> invocation = new RpcInvocation<>(response, context);
+ try {
+ StatusChecker.check(status, invocation);
+ fail();
+ } catch (UnsupportedException ignore) {
+ // ignore on purpose
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImplTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImplTest.java
index 42d881a..7af67d5 100644
--- a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImplTest.java
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImplTest.java
@@ -34,6 +34,7 @@ import apache.rocketmq.v2.Status;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
+import io.grpc.Metadata;
import java.lang.reflect.Field;
import java.time.Duration;
import java.util.ArrayList;
@@ -49,7 +50,9 @@ import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.misc.RequestIdGenerator;
import org.apache.rocketmq.client.java.retry.RetryPolicy;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
+import org.apache.rocketmq.client.java.rpc.Context;
import org.apache.rocketmq.client.java.rpc.RpcInvocation;
+import org.apache.rocketmq.client.java.rpc.Signature;
import org.apache.rocketmq.client.java.tool.TestBase;
import org.junit.Before;
import org.junit.Test;
@@ -154,8 +157,12 @@ public class ProcessQueueImplTest extends TestBase {
List<MessageViewImpl> messageViewList = new ArrayList<>();
final MessageViewImpl messageView = fakeMessageViewImpl();
messageViewList.add(messageView);
- ReceiveMessageResult receiveMessageResult = new ReceiveMessageResult(fakeEndpoints(),
- RequestIdGenerator.getInstance().next(), status, messageViewList);
+ final Metadata metadata = new Metadata();
+ metadata.put(Metadata.Key.of(Signature.REQUEST_ID_KEY, Metadata.ASCII_STRING_MARSHALLER),
+ RequestIdGenerator.getInstance().next());
+ Context context = new Context(fakeEndpoints(), metadata);
+
+ ReceiveMessageResult receiveMessageResult = new ReceiveMessageResult(fakeEndpoints(), messageViewList);
SettableFuture<ReceiveMessageResult> future0 = SettableFuture.create();
future0.set(receiveMessageResult);
when(pushConsumer.receiveMessage(any(ReceiveMessageRequest.class), any(MessageQueueImpl.class),