opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-10-29 17:25:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3949] Add error handle when telemetry

Browse Source
This commit is contained in:
zhouxiang
2022-07-13 11:30:50 +08:00
parent e3658d8a43
commit 3bac7de043
2 changed files with 23 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/GrpcMessagingApplication.java
+5
View File
@@ -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);
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java
+18 -12
View File
@@ -206,19 +206,25 @@ public class ClientActivity extends AbstractMessingActivity {
return new StreamObserver<TelemetryCommand>() {
@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());
}
}