diff --git a/core/trino-main/src/main/java/io/trino/dispatcher/DispatchManager.java b/core/trino-main/src/main/java/io/trino/dispatcher/DispatchManager.java index fd3e9b3563c6..39b7d1795a0d 100644 --- a/core/trino-main/src/main/java/io/trino/dispatcher/DispatchManager.java +++ b/core/trino-main/src/main/java/io/trino/dispatcher/DispatchManager.java @@ -22,6 +22,7 @@ import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.context.Context; import io.trino.Session; +import io.trino.event.QueryMonitor; import io.trino.execution.QueryIdGenerator; import io.trino.execution.QueryInfo; import io.trino.execution.QueryManagerConfig; @@ -56,6 +57,7 @@ import static io.trino.execution.QueryState.RUNNING; import static io.trino.spi.StandardErrorCode.QUERY_TEXT_TOO_LARGE; import static io.trino.tracing.ScopedSpan.scopedSpan; +import static io.trino.util.Failures.toFailure; import static io.trino.util.StatementUtils.getQueryType; import static java.lang.String.format; import static java.util.Objects.requireNonNull; @@ -80,6 +82,7 @@ public class DispatchManager private final QueryTracker queryTracker; private final QueryManagerStats stats = new QueryManagerStats(); + private final QueryMonitor queryMonitor; @Inject public DispatchManager( @@ -94,7 +97,8 @@ public DispatchManager( SessionPropertyManager sessionPropertyManager, Tracer tracer, QueryManagerConfig queryManagerConfig, - DispatchExecutor dispatchExecutor) + DispatchExecutor dispatchExecutor, + QueryMonitor queryMonitor) { this.queryIdGenerator = requireNonNull(queryIdGenerator, "queryIdGenerator is null"); this.queryPreparer = requireNonNull(queryPreparer, "queryPreparer is null"); @@ -112,6 +116,7 @@ public DispatchManager( this.dispatchExecutor = dispatchExecutor.getExecutor(); this.queryTracker = new QueryTracker<>(queryManagerConfig, dispatchExecutor.getScheduledExecutor()); + this.queryMonitor = requireNonNull(queryMonitor, "queryMonitor is null"); } @PostConstruct @@ -236,6 +241,11 @@ private void createQueryInternal(QueryId queryId, Span querySpan, Slug slug, Optional preparedSql = Optional.ofNullable(preparedQuery).flatMap(PreparedQuery::getPrepareSql); DispatchQuery failedDispatchQuery = failedDispatchQueryFactory.createFailedDispatchQuery(session, query, preparedSql, Optional.empty(), throwable); queryCreated(failedDispatchQuery); + // maintain proper order of calls such that EventListener has access to QueryInfo + // - add query to tracker + // - fire query created event + // - fire query completed event + queryMonitor.queryImmediateFailureEvent(failedDispatchQuery.getBasicQueryInfo(), toFailure(throwable)); querySpan.setStatus(StatusCode.ERROR, throwable.getMessage()) .recordException(throwable) .end(); diff --git a/core/trino-main/src/main/java/io/trino/dispatcher/FailedDispatchQueryFactory.java b/core/trino-main/src/main/java/io/trino/dispatcher/FailedDispatchQueryFactory.java index 3eeab6198ef3..65c75d35780e 100644 --- a/core/trino-main/src/main/java/io/trino/dispatcher/FailedDispatchQueryFactory.java +++ b/core/trino-main/src/main/java/io/trino/dispatcher/FailedDispatchQueryFactory.java @@ -24,7 +24,6 @@ import java.util.Optional; import java.util.concurrent.ExecutorService; -import static io.trino.util.Failures.toFailure; import static java.util.Objects.requireNonNull; public class FailedDispatchQueryFactory @@ -58,7 +57,6 @@ public FailedDispatchQuery createFailedDispatchQuery(Session session, String que BasicQueryInfo queryInfo = failedDispatchQuery.getBasicQueryInfo(); queryMonitor.queryCreatedEvent(queryInfo); - queryMonitor.queryImmediateFailureEvent(queryInfo, toFailure(throwable)); return failedDispatchQuery; }