ESQL: Fix async operator warnings not always sent when blocking

Author: ivancea

muted-tests.yml

@@ -286,9 +286,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.action.CrossClusterQueryWithPartialResultsIT
  method: testOneRemoteClusterPartial
  issue: https://github.com/elastic/elasticsearch/issues/124055
-- class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
- method: test {csv-spec:lookup-join.MvJoinKeyOnTheLookupIndex}
- issue: https://github.com/elastic/elasticsearch/issues/128030
 - class: org.elasticsearch.packaging.test.EnrollmentProcessTests
  method: test20DockerAutoFormCluster
  issue: https://github.com/elastic/elasticsearch/issues/128113
@@ -350,9 +347,6 @@ tests:
 - class: org.elasticsearch.index.IndexingPressureIT
  method: testWriteCanRejectOnPrimaryBasedOnMaxOperationSize
  issue: https://github.com/elastic/elasticsearch/issues/130281
-- class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
- method: test {csv-spec:lookup-join.MvJoinKeyOnFrom}
- issue: https://github.com/elastic/elasticsearch/issues/130296
 - class: org.elasticsearch.xpack.esql.inference.bulk.BulkInferenceExecutorTests
  method: testSuccessfulExecution
  issue: https://github.com/elastic/elasticsearch/issues/130306
@@ -362,9 +356,6 @@ tests:
 - class: org.elasticsearch.gradle.LoggedExecFuncTest
  method: failed tasks output logged to console when spooling true
  issue: https://github.com/elastic/elasticsearch/issues/119509
-- class: org.elasticsearch.xpack.esql.qa.single_node.EsqlSpecIT
- method: test {csv-spec:lookup-join.MvJoinKeyFromRow}
- issue: https://github.com/elastic/elasticsearch/issues/130642
 - class: org.elasticsearch.indices.stats.IndexStatsIT
  method: testFilterCacheStats
  issue: https://github.com/elastic/elasticsearch/issues/124447
@@ -374,9 +365,6 @@ tests:
 - class: org.elasticsearch.search.SearchWithRejectionsIT
  method: testOpenContextsAfterRejections
  issue: https://github.com/elastic/elasticsearch/issues/130821
-- class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
- method: test {csv-spec:lookup-join.MvJoinKeyOnFromAfterStats}
- issue: https://github.com/elastic/elasticsearch/issues/131148
 - class: org.elasticsearch.xpack.esql.ccq.MultiClustersIT
  method: testLookupJoinAliases
  issue: https://github.com/elastic/elasticsearch/issues/131166
@@ -488,12 +476,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.inference.completion.CompletionOperatorTests
  method: testSimpleCircuitBreaking
  issue: https://github.com/elastic/elasticsearch/issues/132382
-- class: org.elasticsearch.xpack.esql.ccq.MultiClusterSpecIT
- method: test {csv-spec:lookup-join.MvJoinKeyOnFrom}
- issue: https://github.com/elastic/elasticsearch/issues/132554
-- class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
- method: test {csv-spec:lookup-join.MvJoinKeyFromRow}
- issue: https://github.com/elastic/elasticsearch/issues/132555
 - class: org.elasticsearch.xpack.esql.qa.single_node.EsqlSpecIT
  method: test {csv-spec:spatial.ConvertFromStringParseError}
  issue: https://github.com/elastic/elasticsearch/issues/132558
@@ -506,9 +488,6 @@ tests:
 - class: org.elasticsearch.xpack.logsdb.qa.StoredSourceLogsDbVersusReindexedLogsDbChallengeRestIT
  method: testEsqlSource
  issue: https://github.com/elastic/elasticsearch/issues/132602
-- class: org.elasticsearch.xpack.esql.qa.mixed.MixedClusterEsqlSpecIT
- method: test {csv-spec:lookup-join.MvJoinKeyFromRowExpanded}
- issue: https://github.com/elastic/elasticsearch/issues/132604
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
  method: test {p0=esql/60_usage/Basic ESQL usage output (telemetry) non-snapshot version}
  issue: https://github.com/elastic/elasticsearch/issues/132608

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/AsyncOperator.java

@@ -95,19 +95,16 @@ public void addInput(Page input) {
  driverContext.addAsyncAction();
  boolean success = false;
  try {
- final ActionListener<Fetched> listener = ActionListener.wrap(output -> {
- buffers.put(seqNo, output);
- onSeqNoCompleted(seqNo);
- }, e -> {
+ final ActionListener<Fetched> listener = ActionListener.wrap(output -> buffers.put(seqNo, output), e -> {
  releasePageOnAnyThread(input);
  failureCollector.unwrapAndCollect(e);
- onSeqNoCompleted(seqNo);
  });
  final long startNanos = System.nanoTime();
  performAsync(input, ActionListener.runAfter(listener, () -> {
  responseHeadersCollector.collect();
- driverContext.removeAsyncAction();
  processNanos.add(System.nanoTime() - startNanos);
+ onSeqNoCompleted(seqNo);
+ driverContext.removeAsyncAction();
  }));
  success = true;
  } finally {
@@ -241,7 +238,7 @@ public IsBlockedResult isBlocked() {
 
  @Override
  public final Operator.Status status() {
- return status(Math.max(0L, checkpoint.getMaxSeqNo()), Math.max(0L, checkpoint.getProcessedCheckpoint()), processNanos.sum());
+ return status(checkpoint.getMaxSeqNo() + 1, checkpoint.getProcessedCheckpoint() + 1, processNanos.sum());
  }
 
  protected Operator.Status status(long receivedPages, long completedPages, long processNanos) {
@@ -292,7 +289,7 @@ public long completedPages() {
  return completedPages;
  }
 
- public long procesNanos() {
+ public long processNanos() {
  return processNanos;
  }
 

x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join.csv-spec

@@ -429,7 +429,7 @@ emp_no:integer | language_code:integer | language_name:keyword
 
 mvJoinKeyOnTheLookupIndex
 required_capability: join_lookup_v12
-required_capability: join_lookup_skip_mv_warnings
+required_capability: async_operator_warnings_fix
 
 FROM employees
 | WHERE 10003 < emp_no AND emp_no < 10008
@@ -451,7 +451,7 @@ emp_no:integer | language_code:integer | language_name:keyword
 
 mvJoinKeyOnFrom
 required_capability: join_lookup_v12
-required_capability: join_lookup_skip_mv_warnings
+required_capability: async_operator_warnings_fix
 
 FROM employees
 | WHERE emp_no < 10006
@@ -474,7 +474,7 @@ emp_no:integer | language_code:integer | language_name:keyword
 
 mvJoinKeyOnTheLookupIndexAfterStats
 required_capability: join_lookup_v12
-required_capability: join_lookup_skip_mv_warnings
+required_capability: async_operator_warnings_fix
 
 FROM employees
 | WHERE 10003 < emp_no AND emp_no < 10008
@@ -497,7 +497,7 @@ emp_no:integer | language_code:integer | language_name:keyword
 
 mvJoinKeyOnFromAfterStats
 required_capability: join_lookup_v12
-required_capability: join_lookup_skip_mv_warnings
+required_capability: async_operator_warnings_fix
 
 FROM employees
 | WHERE emp_no < 10006
@@ -521,7 +521,7 @@ emp_no:integer | language_code:integer | language_name:keyword
 
 mvJoinKeyFromRow
 required_capability: join_lookup_v12
-required_capability: join_lookup_skip_mv_warnings
+required_capability: async_operator_warnings_fix
 
 ROW language_code = [4, 5, 6, 7]
 | LOOKUP JOIN languages_lookup_non_unique_key ON language_code
@@ -538,7 +538,7 @@ language_code:integer | language_name:keyword | country:text
 
 mvJoinKeyFromRowExpanded
 required_capability: join_lookup_v12
-required_capability: join_lookup_skip_mv_warnings
+required_capability: async_operator_warnings_fix
 
 ROW language_code = [4, 5, 6, 7, 8]
 | MV_EXPAND language_code

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -825,6 +825,11 @@ public enum Cap {
  */
  JOIN_LOOKUP_SKIP_MV_WARNINGS(JOIN_LOOKUP_V12.isEnabled()),
 
+ /**
+ * Fix for async operator sometimes completing the driver without emitting the stored warnings
+ */
+ ASYNC_OPERATOR_WARNINGS_FIX,
+
  /**
  * Fix pushing down LIMIT past LOOKUP JOIN in case of multiple matching join keys.
  */