lizhimins commented on code in PR #992:
URL: https://github.com/apache/rocketmq-clients/pull/992#discussion_r2076854879
##########
java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java:
##########
@@ -245,47 +247,54 @@ private void receiveMessageImmediately(String attemptId) {
final ListenableFuture<ReceiveMessageResult> future =
consumer.receiveMessage(request, mq,
longPollingTimeout);
- Futures.addCallback(future, new
FutureCallback<ReceiveMessageResult>() {
- @Override
- public void onSuccess(ReceiveMessageResult result) {
- // Intercept after message reception.
- final List<GeneralMessage> generalMessages =
result.getMessageViewImpls().stream()
- .map((Function<MessageView, GeneralMessage>)
GeneralMessageImpl::new)
- .collect(Collectors.toList());
- final MessageInterceptorContextImpl context0 =
- new MessageInterceptorContextImpl(context,
MessageHookPointsStatus.OK);
- consumer.doAfter(context0, generalMessages);
-
- try {
- onReceiveMessageResult(result);
- } catch (Throwable t) {
- // Should never reach here.
- log.error("[Bug] Exception raised while handling
receive result, mq={}, endpoints={}, "
- + "clientId={}", mq, endpoints, clientId, t);
- onReceiveMessageException(t, attemptId);
+ inflightReceiveRequestCount.incrementAndGet();
+ try {
+ Futures.addCallback(future, new
FutureCallback<ReceiveMessageResult>() {
+ @Override
+ public void onSuccess(ReceiveMessageResult result) {
+ // Intercept after message reception.
+ final List<GeneralMessage> generalMessages =
result.getMessageViewImpls().stream()
+ .map((Function<MessageView, GeneralMessage>)
GeneralMessageImpl::new)
+ .collect(Collectors.toList());
+ final MessageInterceptorContextImpl context0 =
+ new MessageInterceptorContextImpl(context,
MessageHookPointsStatus.OK);
+ consumer.doAfter(context0, generalMessages);
+
+ try {
+ onReceiveMessageResult(result);
+ } catch (Throwable t) {
+ // Should never reach here.
+ log.error("[Bug] Exception raised while handling
receive result, mq={}, endpoints={}, "
+ + "clientId={}", mq, endpoints, clientId, t);
+ onReceiveMessageException(t, attemptId);
+ }
}
- }
- @Override
- public void onFailure(Throwable t) {
- String nextAttemptId = null;
- if (t instanceof StatusRuntimeException) {
- StatusRuntimeException exception =
(StatusRuntimeException) t;
- if (io.grpc.Status.DEADLINE_EXCEEDED.getCode() ==
exception.getStatus().getCode()) {
- nextAttemptId = request.getAttemptId();
+ @Override
+ public void onFailure(Throwable t) {
+ String nextAttemptId = null;
+ if (t instanceof StatusRuntimeException) {
+ StatusRuntimeException exception =
(StatusRuntimeException) t;
+ if (io.grpc.Status.DEADLINE_EXCEEDED.getCode() ==
exception.getStatus().getCode()) {
+ nextAttemptId = request.getAttemptId();
+ }
}
+ // Intercept after message reception.
+ final MessageInterceptorContextImpl context0 =
+ new MessageInterceptorContextImpl(context,
MessageHookPointsStatus.ERROR);
+ consumer.doAfter(context0, Collections.emptyList());
+
+ log.error("Exception raised during message reception,
mq={}, endpoints={}, attemptId={}, " +
+ "nextAttemptId={}, clientId={}", mq,
endpoints, request.getAttemptId(), nextAttemptId,
+ clientId, t);
+ onReceiveMessageException(t, nextAttemptId);
Review Comment:
Why not add it to hook
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
