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.
          */