lizhimins commented on code in PR #8430:
URL: https://github.com/apache/rocketmq/pull/8430#discussion_r1694618564
##########
client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java:
##########
@@ -197,37 +191,28 @@ public boolean append(final Object ctx) {
@Override
public void flush() {
- // 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;
- }
- }
+ while (traceContextQueue.size() > 0 || appenderQueue.size() > 0 &&
System.currentTimeMillis() <= end) {
try {
- Thread.sleep(1);
- } catch (InterruptedException e) {
- break;
+ flushTraceContext(true);
+ } catch (Throwable throwable) {
+ log.error("flushTraceContext error", throwable);
}
}
- log.info("------end trace send " + traceContextQueue.size() + " " +
appenderQueue.size());
+ if (appenderQueue.size() > 0) {
+ log.error("There are still some traces that haven't been sent " +
traceContextQueue.size() + " " + appenderQueue.size());
+ }
Review Comment:
这里有并发问题，刚 flush 完就有数据进来了
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
