Gabriel39 commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3849613964
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -147,92 +240,398 @@ public void submitFinishQueryToAudit(AuditEvent event) {
// put the event to queryAuditEventList and let the worker
thread to handle it.
// the worker thread will try best to wait for the statistic
info before logging this event.
event.pushToAuditLogQueueTime = System.currentTimeMillis();
- queryAuditEventList.add(event);
+ PendingAuditEvent pending = new PendingAuditEvent(event,
expectedBackendIds);
+ // Register before publishing to the queue so cleanup cannot
remove a first report
+ // in the gap between queue insertion and participant
retention.
+ registerPendingAuditEvent(pending);
+ queryAuditEventList.add(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
}
- private List<AuditEvent> getQueryNeedAudit() {
- List<AuditEvent> ret = new ArrayList<>();
+ @VisibleForTesting
+ List<AuditEvent> getQueryNeedAudit() {
+ RuntimeStatisticsSnapshot runtimeSnapshot =
buildRuntimeStatisticsSnapshot();
+ queryStatisticsSnapshot =
ImmutableMap.copyOf(runtimeSnapshot.queryStatistics);
+
long currentTime = System.currentTimeMillis();
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ long maximumWaitMs = Math.max(queryAuditLogTimeout,
+ Config.be_report_query_statistics_timeout_ms);
+ List<PendingAuditEvent> dueEvents = new ArrayList<>();
+ // Keep only identity/timestamp traversal under the producer-facing
queue lock; participant
+ // lookup and statistics merging may touch every expected backend and
must run lock-free.
queryAuditEventLogWriteLock();
try {
- int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
- Iterator<AuditEvent> iter = queryAuditEventList.iterator();
- while (iter.hasNext()) {
- AuditEvent ae = iter.next();
- if (currentTime - ae.pushToAuditLogQueueTime >
queryAuditLogTimeout) {
- ret.add(ae);
- iter.remove();
- } else {
+ for (PendingAuditEvent pending : queryAuditEventList) {
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
Review Comment:
Fixed in 7ef7cda086. The audit scan now continues past non-due entries
instead of assuming wall-clock timestamps remain insertion ordered. The
regression test reverses the head and later event timestamps and verifies that
the later due event is emitted immediately.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
