AsyncOperator#isFinished must never return true on failure (elastic#104029) (elastic#104070)

Enrich IT tests can return OK with some missing results instead of
Failure when the enrich lookup hits circuit breakers. This is due to a
race condition in isFinished and onFailure within the AsyncOperator.
When an async lookup fails, we set the exception and then discard pages.
Unfortunately, in the isFinished method, we perform the checks in the
same order: first, we check for failure, and then we check for
outstanding pages. If there is a long pause between these steps,
isFinished might not detect the failure but see no outstanding pages,
leading it to return true despite the presence of a failure. This change
swaps the order of the checks.

Author: dnhatn

docs/changelog/104029.yaml

@@ -0,0 +1,5 @@
+pr: 104029
+summary: '`AsyncOperator#isFinished` must never return true on failure'
+area: ES|QL
+type: bug
+issues: []

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

@@ -181,8 +181,12 @@ public void finish() {
 
     @Override
     public boolean isFinished() {
-        checkFailure();
-        return finished && checkpoint.getPersistedCheckpoint() == checkpoint.getMaxSeqNo();
+        if (finished && checkpoint.getPersistedCheckpoint() == checkpoint.getMaxSeqNo()) {
+            checkFailure();
+            return true;
+        } else {
+            return false;
+        }
     }
 
     @Override

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/AsyncOperatorTests.java

@@ -43,6 +43,8 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.LongStream;
@@ -270,6 +272,53 @@ protected void doClose() {
         }
     }
 
+    public void testIsFinished() {
+        int iters = iterations(10, 10_000);
+        BlockFactory blockFactory = blockFactory();
+        for (int i = 0; i < iters; i++) {
+            DriverContext driverContext = new DriverContext(blockFactory.bigArrays(), blockFactory);
+            CyclicBarrier barrier = new CyclicBarrier(2);
+            AsyncOperator asyncOperator = new AsyncOperator(driverContext, between(1, 10)) {
+                @Override
+                protected void performAsync(Page inputPage, ActionListener<Page> listener) {
+                    ActionRunnable<Page> command = new ActionRunnable<>(listener) {
+                        @Override
+                        protected void doRun() {
+                            try {
+                                barrier.await(10, TimeUnit.SECONDS);
+                            } catch (Exception e) {
+                                throw new AssertionError(e);
+                            }
+                            listener.onFailure(new ElasticsearchException("simulated"));
+                        }
+                    };
+                    threadPool.executor(ESQL_TEST_EXECUTOR).execute(command);
+                }
+
+                @Override
+                protected void doClose() {
+
+                }
+            };
+            asyncOperator.addInput(new Page(blockFactory.newConstantIntBlockWith(randomInt(), between(1, 10))));
+            asyncOperator.finish();
+            try {
+                barrier.await(10, TimeUnit.SECONDS);
+            } catch (Exception e) {
+                throw new AssertionError(e);
+            }
+            int numChecks = between(10, 100);
+            while (--numChecks >= 0) {
+                try {
+                    assertFalse("must not finished or failed", asyncOperator.isFinished());
+                } catch (ElasticsearchException e) {
+                    assertThat(e.getMessage(), equalTo("simulated"));
+                    break;
+                }
+            }
+        }
+    }
+
     static class LookupService {
         private final ThreadPool threadPool;
         private final Map<Long, String> dict;