diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java b/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java index 96043db95c..139d7232f7 100644 --- a/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java +++ b/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java @@ -99,12 +99,12 @@ public class AsyncTraceDispatcher implements TraceDispatcher { this.traceTopicName = TopicValidator.RMQ_SYS_TRACE_TOPIC; } this.traceExecutor = new ThreadPoolExecutor(// - 10, // - 20, // - 1000 * 60, // - TimeUnit.MILLISECONDS, // - this.appenderQueue, // - new ThreadFactoryImpl("MQTraceSendThread_")); + 10, // + 20, // + 1000 * 60, // + TimeUnit.MILLISECONDS, // + this.appenderQueue, // + new ThreadFactoryImpl("MQTraceSendThread_")); traceProducer = getAndCreateTraceProducer(rpcHook); } @@ -188,6 +188,11 @@ public class AsyncTraceDispatcher implements TraceDispatcher { // The maximum waiting time for refresh,avoid being written all the time, resulting in failure to return. long end = System.currentTimeMillis() + 500; while (System.currentTimeMillis() <= end) { + synchronized (taskQueueByTopic) { + for (TraceDataSegment taskInfo : taskQueueByTopic.values()) { + taskInfo.sendAllData(); + } + } synchronized (traceContextQueue) { if (traceContextQueue.size() == 0 && appenderQueue.size() == 0) { break; @@ -252,7 +257,7 @@ public class AsyncTraceDispatcher implements TraceDispatcher { while (System.currentTimeMillis() < endTime) { try { TraceContext traceContext = traceContextQueue.poll( - endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS + endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS ); if (traceContext != null && !traceContext.getTraceBeans().isEmpty()) {