opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 15:47:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #9377] fix 'send trace data can fail if shutdown producer immediately' (#9378)

Browse Source
This commit is contained in:
kaikoo
2025-04-30 15:15:59 +08:00
committed by GitHub
parent ade18bb076
commit c9deaa538b
1 changed files with 4 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java
+4 -6
View File
@@ -44,6 +44,7 @@ import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.RPCHook;
@@ -54,6 +55,7 @@ public class AsyncTraceDispatcher implements TraceDispatcher {
private static final Logger log = LoggerFactory.getLogger(AsyncTraceDispatcher.class);
private static final AtomicInteger COUNTER = new AtomicInteger();
private static final AtomicInteger INSTANCE_NUM = new AtomicInteger(0);
private static final long WAIT_FOR_SHUTDOWN = 5000L;
private volatile boolean stopped = false;
private final int traceInstanceId = INSTANCE_NUM.getAndIncrement();
private final int batchNum;
@@ -190,23 +192,19 @@ public class AsyncTraceDispatcher implements TraceDispatcher {
@Override
public void flush() {
long end = System.currentTimeMillis() + 500;
while (traceContextQueue.size() > 0 || appenderQueue.size() > 0 && System.currentTimeMillis() <= end) {
while (traceContextQueue.size() > 0) {
try {
flushTraceContext(true);
} catch (Throwable throwable) {
log.error("flushTraceContext error", throwable);
}
}
if (appenderQueue.size() > 0) {
log.error("There are still some traces that haven't been sent " + traceContextQueue.size() + " " + appenderQueue.size());
}
}
@Override
public void shutdown() {
flush();
this.traceExecutor.shutdown();
ThreadUtils.shutdownGracefully(this.traceExecutor, WAIT_FOR_SHUTDOWN, TimeUnit.MILLISECONDS);
if (isStarted.get()) {
traceProducer.shutdown();
}