github-actions[bot] commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3849139567
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -480,7 +500,13 @@ void
RuntimeQueryStatisticsMgr::report_runtime_query_statistics() {
bool is_query_finished = qs_status_pair.first;
bool is_timeout_after_finish = qs_status_pair.second;
if ((is_rpc_success && is_query_finished) ||
is_timeout_after_finish) {
- _resource_contexts_map.erase(query_id);
+ auto iter = _resource_contexts_map.find(query_id);
+ // A later group-commit generation may reuse the query ID
while this RPC is
+ // in flight; only erase the exact ResourceContext that
was reported.
+ if (iter != _resource_contexts_map.end() &&
+ iter->second == qs_resource_contexts[query_id]) {
Review Comment:
[P1] Keep rejected final snapshots for the full FE fallback window. A
heartbeat-lagging FE correctly returns
`workload_runtime_status_accepted=false`, but the timeout disjunct above still
reaches this erase after the BE default of 30 seconds, while FE now waits 60
seconds before giving up on the audit. If heartbeat catches up at t=40, the
only final `ResourceContext` is already gone and the audit must log without its
counters. Preserve a compact terminal snapshot until acceptance or at least
align retention with FE's maximum wait, and add a
reject-past-30s/catch-up-before-60s test.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -147,92 +182,261 @@ 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,
+ bindAndRetainBackendIncarnations(event.queryId,
expectedBackendIds));
+ queryAuditEventList.add(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
}
- private List<AuditEvent> getQueryNeedAudit() {
+ @VisibleForTesting
+ List<AuditEvent> getQueryNeedAudit() {
+ RuntimeStatisticsSnapshot runtimeSnapshot =
buildRuntimeStatisticsSnapshot();
+ queryStatisticsSnapshot =
ImmutableMap.copyOf(runtimeSnapshot.queryStatistics);
+
List<AuditEvent> ret = new ArrayList<>();
long currentTime = System.currentTimeMillis();
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ long maximumWaitMs = Math.max(queryAuditLogTimeout,
+ Config.be_report_query_statistics_timeout_ms);
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 {
+ Iterator<PendingAuditEvent> iterator =
queryAuditEventList.iterator();
+ while (iterator.hasNext()) {
+ PendingAuditEvent pending = iterator.next();
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
+ // Events are appended in timestamp order, so later
entries cannot be due yet.
break;
}
+ boolean allFinalSnapshotsReceived = true;
+ TQueryStatistics auditStatistics =
pending.expectedBackendIds.length == 0
+ ?
runtimeSnapshot.queryStatistics.get(pending.event.queryId)
+ : new TQueryStatistics();
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
+ long backendStartTime =
pending.expectedBackendStartTimes[i];
+ if (backendStartTime == 0) {
Review Comment:
[P1] Bind a pending participant when its first matching report arrives. If
q1 is enqueued before process A's periodic report, this start time stays zero;
A's final snapshot can then be accepted, but a restart before the five-second
drain makes this code substitute process B and ignore A's retained final data.
This is the inverse ordering of the existing report-before-enqueue test.
Propagate the scheduled incarnation or atomically bind `(queryId, backendId)`
on its first accepted report, and cover enqueue -> final A -> restart -> drain.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -147,92 +182,261 @@ 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,
+ bindAndRetainBackendIncarnations(event.queryId,
expectedBackendIds));
+ queryAuditEventList.add(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
}
- private List<AuditEvent> getQueryNeedAudit() {
+ @VisibleForTesting
+ List<AuditEvent> getQueryNeedAudit() {
+ RuntimeStatisticsSnapshot runtimeSnapshot =
buildRuntimeStatisticsSnapshot();
+ queryStatisticsSnapshot =
ImmutableMap.copyOf(runtimeSnapshot.queryStatistics);
+
List<AuditEvent> ret = new ArrayList<>();
long currentTime = System.currentTimeMillis();
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ long maximumWaitMs = Math.max(queryAuditLogTimeout,
+ Config.be_report_query_statistics_timeout_ms);
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 {
+ Iterator<PendingAuditEvent> iterator =
queryAuditEventList.iterator();
+ while (iterator.hasNext()) {
+ PendingAuditEvent pending = iterator.next();
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
+ // Events are appended in timestamp order, so later
entries cannot be due yet.
break;
}
+ boolean allFinalSnapshotsReceived = true;
+ TQueryStatistics auditStatistics =
pending.expectedBackendIds.length == 0
+ ?
runtimeSnapshot.queryStatistics.get(pending.event.queryId)
+ : new TQueryStatistics();
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
+ long backendStartTime =
pending.expectedBackendStartTimes[i];
+ if (backendStartTime == 0) {
+ backendStartTime =
getBackendStartTime(pending.expectedBackendIds[i]);
+ }
+ TQueryStatisticsResult statistics =
findStatisticsForBackend(
+ pending.event.queryId,
pending.expectedBackendIds[i], backendStartTime);
+ if (statistics == null) {
+ allFinalSnapshotsReceived = false;
+ continue;
+ }
+ mergeQueryStatistics(auditStatistics, statistics);
+ if (!statistics.isSetQueryFinished() ||
!statistics.isQueryFinished()) {
+ allFinalSnapshotsReceived = false;
+ }
+ }
+ if (!allFinalSnapshotsReceived && elapsed <= maximumWaitMs) {
+ continue;
+ }
+ // A DML audit must not pass the queue until every scheduled
BE has published its
+ // query-level final snapshot. The upper bound only protects
the queue if a BE dies.
+ applyQueryStatisticsToAuditEvent(pending.event,
auditStatistics);
+ ret.add(pending.event);
+ iterator.remove();
+ releaseBoundIncarnations(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
return ret;
}
- public void updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
+ public boolean updateBeQueryStats(TReportWorkloadRuntimeStatusParams
params) {
if (!params.isSetBackendId()) {
LOG.warn("be report workload runtime status but without beid");
- return;
+ return false;
}
if (!params.isSetQueryStatisticsResultMap()) {
LOG.warn("be report workload runtime status but without query
stats map");
- return;
+ return false;
+ }
+ if (!params.isSetBackendStartTime()) {
+ LOG.warn("be report workload runtime status without backend start
time");
+ return false;
}
long beId = params.backend_id;
- // NOTE(wb) one be sends update request one by one,
- // so there is no need a global lock for beToQueryStatsMap here,
- // just keep one be's put/remove/get is atomic operation is enough
- long currentTime = System.currentTimeMillis();
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
- if (beReportInfo == null) {
- beReportInfo = new BeReportInfo(currentTime);
- beToQueryStatsMap.put(beId, beReportInfo);
+ long backendStartTime = params.backend_start_time;
+ long currentBackendStartTime = getBackendStartTime(beId);
+ if (currentBackendStartTime > 0 && currentBackendStartTime !=
backendStartTime) {
+ LOG.info("ignore stale workload runtime status from backend {},
report start time {}, current {}",
+ beId, backendStartTime, currentBackendStartTime);
+ return false;
+ }
+ if (currentBackendStartTime == 0) {
Review Comment:
[P2] Treat every non-positive heartbeat start time as unknown. A follower
can hold a newly added backend at the production `-1` sentinel while the master
has already heartbeated it and schedules forwarded DML. This `== 0` guard is
then skipped, so if the BE restarts again before heartbeat replay catches up,
old/new process reports overwrite the latch and can both be acknowledged,
allowing both sides to discard retry state. Use `<= 0` for the first-observed
guard and test competing report epochs while the audit FE reports `-1`.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -147,92 +182,261 @@ 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,
+ bindAndRetainBackendIncarnations(event.queryId,
expectedBackendIds));
+ queryAuditEventList.add(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
}
- private List<AuditEvent> getQueryNeedAudit() {
+ @VisibleForTesting
+ List<AuditEvent> getQueryNeedAudit() {
+ RuntimeStatisticsSnapshot runtimeSnapshot =
buildRuntimeStatisticsSnapshot();
+ queryStatisticsSnapshot =
ImmutableMap.copyOf(runtimeSnapshot.queryStatistics);
+
List<AuditEvent> ret = new ArrayList<>();
long currentTime = System.currentTimeMillis();
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ long maximumWaitMs = Math.max(queryAuditLogTimeout,
+ Config.be_report_query_statistics_timeout_ms);
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 {
+ Iterator<PendingAuditEvent> iterator =
queryAuditEventList.iterator();
+ while (iterator.hasNext()) {
+ PendingAuditEvent pending = iterator.next();
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
+ // Events are appended in timestamp order, so later
entries cannot be due yet.
break;
}
+ boolean allFinalSnapshotsReceived = true;
+ TQueryStatistics auditStatistics =
pending.expectedBackendIds.length == 0
+ ?
runtimeSnapshot.queryStatistics.get(pending.event.queryId)
+ : new TQueryStatistics();
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
+ long backendStartTime =
pending.expectedBackendStartTimes[i];
+ if (backendStartTime == 0) {
+ backendStartTime =
getBackendStartTime(pending.expectedBackendIds[i]);
+ }
+ TQueryStatisticsResult statistics =
findStatisticsForBackend(
+ pending.event.queryId,
pending.expectedBackendIds[i], backendStartTime);
+ if (statistics == null) {
+ allFinalSnapshotsReceived = false;
+ continue;
+ }
+ mergeQueryStatistics(auditStatistics, statistics);
+ if (!statistics.isSetQueryFinished() ||
!statistics.isQueryFinished()) {
+ allFinalSnapshotsReceived = false;
+ }
+ }
+ if (!allFinalSnapshotsReceived && elapsed <= maximumWaitMs) {
+ continue;
+ }
+ // A DML audit must not pass the queue until every scheduled
BE has published its
+ // query-level final snapshot. The upper bound only protects
the queue if a BE dies.
+ applyQueryStatisticsToAuditEvent(pending.event,
auditStatistics);
+ ret.add(pending.event);
+ iterator.remove();
+ releaseBoundIncarnations(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
return ret;
}
- public void updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
+ public boolean updateBeQueryStats(TReportWorkloadRuntimeStatusParams
params) {
if (!params.isSetBackendId()) {
LOG.warn("be report workload runtime status but without beid");
- return;
+ return false;
}
if (!params.isSetQueryStatisticsResultMap()) {
LOG.warn("be report workload runtime status but without query
stats map");
- return;
+ return false;
+ }
+ if (!params.isSetBackendStartTime()) {
+ LOG.warn("be report workload runtime status without backend start
time");
+ return false;
}
long beId = params.backend_id;
- // NOTE(wb) one be sends update request one by one,
- // so there is no need a global lock for beToQueryStatsMap here,
- // just keep one be's put/remove/get is atomic operation is enough
- long currentTime = System.currentTimeMillis();
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
- if (beReportInfo == null) {
- beReportInfo = new BeReportInfo(currentTime);
- beToQueryStatsMap.put(beId, beReportInfo);
+ long backendStartTime = params.backend_start_time;
+ long currentBackendStartTime = getBackendStartTime(beId);
+ if (currentBackendStartTime > 0 && currentBackendStartTime !=
backendStartTime) {
+ LOG.info("ignore stale workload runtime status from backend {},
report start time {}, current {}",
+ beId, backendStartTime, currentBackendStartTime);
+ return false;
+ }
+ if (currentBackendStartTime == 0) {
+ Long acceptedStartTime =
lastAcceptedBackendStartTimes.putIfAbsent(beId, backendStartTime);
+ if (acceptedStartTime != null && acceptedStartTime !=
backendStartTime) {
+ return false;
+ }
} else {
- beReportInfo.beLastReportTime = currentTime;
+ lastAcceptedBackendStartTimes.put(beId, backendStartTime);
}
- for (Map.Entry<String, TQueryStatisticsResult> entry :
params.query_statistics_result_map.entrySet()) {
- beReportInfo.queryStatsMap.put(entry.getKey(),
Pair.of(currentTime, entry.getValue()));
+ BackendIncarnation incarnation = new BackendIncarnation(beId,
backendStartTime);
+ beToQueryStatsMap.computeIfAbsent(incarnation, ignored -> new
BeReportInfo(backendStartTime));
+ long currentTime = System.currentTimeMillis();
+ for (Map.Entry<String, TQueryStatisticsResult> entry
+ : params.query_statistics_result_map.entrySet()) {
+ beforeUpdateQueryStatistics(entry.getKey());
+ while (true) {
+ BeReportInfo beReportInfo =
beToQueryStatsMap.computeIfAbsent(incarnation,
+ ignored -> new BeReportInfo(backendStartTime));
+ beReportInfo.lifecycleLock.readLock().lock();
+ try {
+ // Cleanup removes only an empty shell under the write
lock. If this insertion
+ // raced with removal, retry the single entry against the
replacement shell.
+ if (beToQueryStatsMap.get(incarnation) != beReportInfo) {
+ continue;
+ }
+ Pair<Long, TQueryStatisticsResult> incoming =
+ Pair.of(currentTime, entry.getValue());
+ beReportInfo.queryStatsMap.compute(entry.getKey(),
(queryId, previous) ->
+ previous == null
+ ||
shouldReplaceQueryStatistics(previous.second, incoming.second)
+ ? incoming : previous);
+ break;
+ } finally {
+ beReportInfo.lifecycleLock.readLock().unlock();
+ }
+ }
+ }
+ // Heartbeat can change while a large report is being merged. Refuse
the acknowledgement
+ // so the BE retries against the FE state that is authoritative after
the merge.
+ long latestBackendStartTime = getBackendStartTime(beId);
+ return latestBackendStartTime <= 0 || latestBackendStartTime ==
backendStartTime;
+ }
+
+ private TQueryStatisticsResult findStatisticsForBackend(String queryId,
+ long backendId, long backendStartTime) {
+ long reportStartTime = backendStartTime > 0
+ ? backendStartTime :
lastAcceptedBackendStartTimes.getOrDefault(backendId, 0L);
+ BeReportInfo reportInfo = beToQueryStatsMap.get(
+ new BackendIncarnation(backendId, reportStartTime));
+ if (reportInfo != null) {
+ Pair<Long, TQueryStatisticsResult> pair =
reportInfo.queryStatsMap.get(queryId);
+ return pair == null ? null : pair.second;
}
+ return null;
}
- private void clearReportTimeoutBeStatistics() {
- // 1 clear report timeout be
- Set<Long> currentBeIdSet = beToQueryStatsMap.keySet();
+ private Map<Long, Long> bindAndRetainBackendIncarnations(String queryId,
Set<Long> backendIds) {
+ Map<Long, Long> result = new HashMap<>();
+ for (long backendId : backendIds) {
+ long currentBackendStartTime = getBackendStartTime(backendId);
+ long reportStartTime = currentBackendStartTime > 0
+ ? currentBackendStartTime
+ : lastAcceptedBackendStartTimes.getOrDefault(backendId,
0L);
+ BackendIncarnation incarnation = new BackendIncarnation(backendId,
reportStartTime);
+ if (reportStartTime == 0) {
+ continue;
+ }
+ retainIncarnation(incarnation);
+ BeReportInfo reportInfo = beToQueryStatsMap.get(incarnation);
+ // Bind only after this FE has observed the query from that
process. A forwarding FE
+ // can lag the coordinator FE's heartbeat and must not bind a new
query to the old BE.
+ if (reportInfo != null &&
reportInfo.queryStatsMap.containsKey(queryId)) {
+ result.put(backendId, reportStartTime);
+ } else {
+ releaseIncarnation(incarnation);
+ }
+ }
+ return result;
+ }
+
+ private void releaseBoundIncarnations(PendingAuditEvent pending) {
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
+ if (pending.expectedBackendStartTimes[i] > 0) {
+ BackendIncarnation incarnation = new BackendIncarnation(
+ pending.expectedBackendIds[i],
pending.expectedBackendStartTimes[i]);
+ releaseIncarnation(incarnation);
+ }
+ }
+ }
+
+ private void retainIncarnation(BackendIncarnation incarnation) {
+ // Retain and release must mutate the map entry atomically;
incrementing a detached counter
+ // could otherwise lose the reference while the last concurrent audit
is being released.
+ pendingIncarnationReferences.compute(incarnation,
+ (ignored, count) -> count == null ? 1 : count + 1);
+ }
+
+ private void releaseIncarnation(BackendIncarnation incarnation) {
+ pendingIncarnationReferences.computeIfPresent(incarnation, (ignored,
count) ->
+ count == 1 ? null : count - 1);
+ }
+
+ @VisibleForTesting
+ long getBackendStartTime(long backendId) {
+ Backend backend = Env.getCurrentSystemInfo().getBackend(backendId);
+ return backend == null ? 0 : backend.getLastStartTime();
+ }
+
+ @VisibleForTesting
+ void beforeUpdateQueryStatistics(String queryId) {
+ }
+
+ private boolean shouldReplaceQueryStatistics(TQueryStatisticsResult
previous,
+ TQueryStatisticsResult incoming) {
+ long previousGeneration = previous.isSetQueryStatisticsGeneration()
+ ? previous.getQueryStatisticsGeneration() : 0;
+ long incomingGeneration = incoming.isSetQueryStatisticsGeneration()
+ ? incoming.getQueryStatisticsGeneration() : 0;
+ if (previousGeneration != incomingGeneration) {
+ return incomingGeneration > previousGeneration;
+ }
+
+ long previousSequence = previous.isSetQueryStatisticsSequence()
+ ? previous.getQueryStatisticsSequence() : 0;
+ long incomingSequence = incoming.isSetQueryStatisticsSequence()
+ ? incoming.getQueryStatisticsSequence() : 0;
+ if (previousSequence != incomingSequence) {
+ return incomingSequence > previousSequence;
+ }
+
+ return false;
+ }
+
+ void clearReportTimeoutBeStatistics() {
+ Set<BackendIncarnation> currentBeIdSet = beToQueryStatsMap.keySet();
Long currentTime = System.currentTimeMillis();
- for (Long beId : currentBeIdSet) {
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
- if (currentTime - beReportInfo.beLastReportTime >
Config.be_report_query_statistics_timeout_ms) {
- beToQueryStatsMap.remove(beId);
+ for (BackendIncarnation incarnation : currentBeIdSet) {
+ BeReportInfo beReportInfo = beToQueryStatsMap.get(incarnation);
+ if (beReportInfo == null) {
continue;
}
Set<String> queryIdSet = beReportInfo.queryStatsMap.keySet();
for (String queryId : queryIdSet) {
- Pair<Long, TQueryStatisticsResult> pair =
beReportInfo.queryStatsMap.get(queryId);
- long queryLastReportTime = pair.first;
- boolean timeout = currentTime - queryLastReportTime
- > Config.be_report_query_statistics_timeout_ms;
- // Remove query statistics only when both conditions are
satisfied:
- // 1) this query statistics is timeout, and
- // 2) FE no longer has this query in QeProcessorImpl.
- // Example timeline:
- // - t0: query q1 is still running, but one periodic BE report
is delayed for > timeout.
- // - t1: clear thread runs. timeout condition is true, but q1
still exists in FE.
- // - t2: we keep q1 statistics instead of removing it; later
reports can update it again.
- if (timeout && isQueryNotExistInFe(queryId)) {
- beReportInfo.queryStatsMap.remove(queryId);
+ beforeExpireQueryStatistics(queryId);
+ beReportInfo.queryStatsMap.computeIfPresent(queryId,
(ignoredQueryId, pair) -> {
+ long queryLastReportTime = pair.first;
+ boolean timeout = currentTime - queryLastReportTime
+ > Config.be_report_query_statistics_timeout_ms;
+ // Conditional removal and updates serialize only for the
same query. Work on
+ // a high-cardinality report cannot delay another query's
terminal update.
+ return timeout &&
!pendingIncarnationReferences.containsKey(incarnation)
Review Comment:
[P1] Make audit retention query-scoped. Each queued DML increments one
reference for the whole BE incarnation, and this condition prevents expiry of
every query in that process while any audit is pending. Because events stay
queued at least five seconds and the daemon runs every two seconds, sustained
inserts can keep the reference nonzero indefinitely, so the 60-second cleaner
never removes completed entries and FE memory grows with total query count.
Track `(incarnation, queryId)` (or retain the exact entry) and test expiration
of q0 while overlapping audits for other queries remain queued.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -147,92 +182,261 @@ 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,
+ bindAndRetainBackendIncarnations(event.queryId,
expectedBackendIds));
+ queryAuditEventList.add(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
}
- private List<AuditEvent> getQueryNeedAudit() {
+ @VisibleForTesting
+ List<AuditEvent> getQueryNeedAudit() {
+ RuntimeStatisticsSnapshot runtimeSnapshot =
buildRuntimeStatisticsSnapshot();
+ queryStatisticsSnapshot =
ImmutableMap.copyOf(runtimeSnapshot.queryStatistics);
+
List<AuditEvent> ret = new ArrayList<>();
long currentTime = System.currentTimeMillis();
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ long maximumWaitMs = Math.max(queryAuditLogTimeout,
+ Config.be_report_query_statistics_timeout_ms);
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 {
+ Iterator<PendingAuditEvent> iterator =
queryAuditEventList.iterator();
+ while (iterator.hasNext()) {
+ PendingAuditEvent pending = iterator.next();
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
+ // Events are appended in timestamp order, so later
entries cannot be due yet.
break;
}
+ boolean allFinalSnapshotsReceived = true;
+ TQueryStatistics auditStatistics =
pending.expectedBackendIds.length == 0
+ ?
runtimeSnapshot.queryStatistics.get(pending.event.queryId)
+ : new TQueryStatistics();
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
Review Comment:
[P2] Move participant hydration out of the audit-queue lock. This critical
section now performs heartbeat lookups, map reads, and statistics merges for
every expected backend of every due event, and repeats that work every two
seconds for unfinished events. With a queue limit of 250,000 and a 60-second
wait, statement threads calling `submitFinishQueryToAudit()` can stall behind
an O(events * participants) daemon pass. Snapshot immutable due entries under
the lock, hydrate them outside it, then reacquire briefly for conditional
removal/release; add a latch test proving submissions proceed while hydration
is blocked.
--
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]
