Gabriel39 commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3849276119
##########
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:
Fixed. FE now returns its live audit fallback window in the workload report
response. After the local retention period, BE converts the finished
ResourceContext into a compact terminal snapshot and keeps retrying it until FE
acceptance or the FE-provided window expires.
RejectedFinalSnapshotUsesFeAuditRetentionWindow covers rejection at 40 seconds
and acceptance after heartbeat catch-up at 50 seconds.
##########
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:
Fixed. Pending references are now keyed by backend incarnation and query ID,
while an unbound participant temporarily protects only the matching backend ID
and query ID candidate. testPendingAuditRetainsOnlyItsOwnQuery verifies that q0
expires while q1 from the same process remains retained.
##########
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:
Fixed. Pending participants are indexed by backend ID and query ID and
atomically bind to the first accepted report epoch. The cleanup path is
serialized with first-report binding to close the insert-to-bind race.
testAuditBindsFirstFinalReportBeforeBackendRestart covers enqueue -> final A ->
restart -> drain, and testFirstReportSurvivesCleanupBeforeParticipantBinding
covers concurrent cleanup.
##########
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:
Fixed. Every non-positive heartbeat start time is now treated as unknown, so
the first accepted report latches the process epoch for both 0 and the
production -1 sentinel. testUnknownHeartbeatAcceptsOnlyFirstReportedIncarnation
verifies that a competing epoch is rejected.
##########
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:
Fixed. The daemon snapshots only due event identities under the audit queue
lock, performs participant binding, lookup, and statistics merging outside it,
then reacquires the lock only for identity-based removals.
testAuditSubmissionDoesNotWaitForParticipantHydration blocks hydration with a
latch and verifies that a concurrent submission completes.
--
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]
