From 3bac7de04399c7498afdba7c58f0e3d3156b4033 Mon Sep 17 00:00:00 2001 From: zhouxiang Date: Tue, 31 May 2022 20:37:05 +0800 Subject: [PATCH] [ISSUE #3949] Add error handle when telemetry --- .../grpc/v2/GrpcMessagingApplication.java | 5 ++++ .../proxy/grpc/v2/client/ClientActivity.java | 30 +++++++++++-------- 2 files changed, 23 insertions(+), 12 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/GrpcMessagingApplication.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/GrpcMessagingApplication.java index 52cea43719..3538fd2b94 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/GrpcMessagingApplication.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/GrpcMessagingApplication.java @@ -48,6 +48,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.thread.ThreadPoolMonitor; import org.apache.rocketmq.proxy.common.StartAndShutdown; import org.apache.rocketmq.proxy.config.ConfigurationManager; @@ -55,8 +56,11 @@ import org.apache.rocketmq.proxy.config.ProxyConfig; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseWriter; import org.apache.rocketmq.proxy.processor.MessagingProcessor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class GrpcMessagingApplication extends MessagingServiceGrpc.MessagingServiceImplBase implements StartAndShutdown { + private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); private final GrpcMessingActivity grpcMessingActivity; @@ -345,6 +349,7 @@ public class GrpcMessagingApplication extends MessagingServiceGrpc.MessagingServ @Override public void onError(Throwable t) { + log.error("telemetry onError", t); responseTelemetryCommand.onError(t); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java index 6ea8f5450e..86b9652c66 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java @@ -206,19 +206,25 @@ public class ClientActivity extends AbstractMessingActivity { return new StreamObserver() { @Override public void onNext(TelemetryCommand request) { - switch (request.getCommandCase()) { - case SETTINGS: { - responseObserver.onNext(processClientSettings(ctx, request, responseObserver)); - break; - } - case THREAD_STACK_TRACE: { - reportThreadStackTrace(ctx, request.getStatus(), request.getThreadStackTrace()); - break; - } - case VERIFY_MESSAGE_RESULT: { - reportVerifyMessageResult(ctx, request.getStatus(), request.getVerifyMessageResult()); - break; + try { + switch (request.getCommandCase()) { + case SETTINGS: { + responseObserver.onNext(processClientSettings(ctx, request, responseObserver)); + break; + } + case THREAD_STACK_TRACE: { + reportThreadStackTrace(ctx, request.getStatus(), request.getThreadStackTrace()); + break; + } + case VERIFY_MESSAGE_RESULT: { + reportVerifyMessageResult(ctx, request.getStatus(), request.getVerifyMessageResult()); + break; + } } + } catch (Exception e) { + responseObserver.onNext(TelemetryCommand.newBuilder() + .setStatus(ResponseBuilder.buildStatus(e)) + .build()); } }