ESQL: Fix async operator warnings not always sent when blocking

[ivancea]

Fixes #128030
Fixes #130296
Fixes #130642
Fixes #131148
Fixes #132554
Fixes #132555
Fixes #132604
Fixes #131563
Fixes #132778

Extracted from #132738

An AsyncOperator listener misordering caused the warnings collection and status metrics updates to be executed after the onSeqNoCompleted()>notifyIfBlocked()>future.onResponse(null), which ends the processing in some cases.

Bug explanation

An example profile of one of the CI fails:

{ planning={took_millis=8, start_millis=1754605547195, stop_millis=1754605547203, took_nanos=8612200}, plans=[ { cluster_name=test-cluster, node_name=test-cluster-1, description=single, plan= OutputExec[org.elasticsearch.xpack.esql.plugin.ComputeService$$Lambda/0x0000000095705068@4387ca3] \_ProjectExec[[language_code{r}#25708, language_name{f}#25719, country{f}#25716]] \_TopNExec[[Order[language_code{r}#25708,ASC,LAST], Order[language_name{f}#25719,ASC,LAST], Order[country{f}#25716,ASC,LAST]],1000[INTEGER],104] \_LookupJoinExec[[language_code{r}#25708],[language_code{f}#25718],[country{f}#25716, language_name{f}#25719]] |_LocalSourceExec[[language_code{r}#25708],org.elasticsearch.xpack.esql.plan.logical.local.CopyingLocalSupplier@1ef1bc4] \_EsQueryExec[languages_lookup_non_unique_key], indexMode[lookup], query[][_doc{f}#25720], limit[], sort[] estimatedRowSize[104] } ], query={took_millis=14, start_millis=1754605547195, stop_millis=1754605547209, took_nanos=14013972}, drivers=[ { cluster_name=test-cluster, sleeps={last=[], counts={}, first=[]}, cpu_nanos=3189539, operators=[ {operator=LocalSourceOperator}, { operator=LookupOperator[ index=languages_lookup_non_unique_key input_type=INTEGER match_field=language_code load_fields=[country{f}#25716, language_name{f}#25719] inputChannel=0 ], status={received_pages=0, total_terms=4, emitted_pages=1, process_nanos=0, completed_pages=0} }, { operator=TopNOperator[ count=0/1000, elementTypes=[INT, BYTES_REF, BYTES_REF], encoders=[DefaultSortable, UTF8TopNEncoder, UTF8TopNEncoder], sortOrders=[SortOrder[channel=0, asc=true, nullsFirst=false], SortOrder[channel=2, asc=true, nullsFirst=false], SortOrder[channel=1, asc=true, nullsFirst=false]] ], status={receive_nanos=36566, emit_nanos=33132, occupied_rows=0, rows_emitted=1, pages_emitted=1, ram_used=4.2kb, rows_received=1, pages_received=1, ram_bytes_used=4368} }, {operator=ProjectOperator[projection = [0, 2, 1]], status={pages_processed=1, rows_emitted=1, process_nanos=2856, rows_received=1}}, {operator=OutputOperator[columns = [language_code, language_name, country]]} ], node_name=test-cluster-1, description=single, documents_found=0, start_millis=1754605547205, stop_millis=1754605547208, values_loaded=0, iterations=3124, took_nanos=3432530 } ] } 

The LookupOperator has this strange status:
status={received_pages=0, total_terms=4, emitted_pages=1, process_nanos=0, completed_pages=0}

0 received pages, 0 completed pages, 0 prcess_nanos, but 4 total terms and 1 emitted page. All the 0s come from AsyncOperator, while the others come from LookupFromIndexOperator.

After some investigation, the received_pages and completed_pages were actually wrong (Fixed in #132738), but the process_nanos shows the issue: That part of the code didn't get to execute yet.

[elasticsearchmachine]

Pinging @elastic/es-analytical-engine (Team:Analytics)

[elasticsearchmachine]

Hi @ivancea, I've created a changelog YAML for you.

[elasticsearchmachine]

Hi @ivancea, I've updated the changelog YAML for you.

[nik9000]

Looks better to me. Let @dnhatn look too please.

[nik9000]

@dnhatn , could you have a look at this bit?

[nik9000]

Actually, this whole PR to be honest. It's the kind of thing I don't trust myself with. Obviously, I broke it in the first place.

[ivancea]

Hey! This change feels a bit mystic. It's properly tested in the other PR, where I fix and test multiple things around the status() here.

I'll remove it from here, it doesn't make sense in this PR, and I wanted only the other bugfix to be here!

[dnhatn]

Thanks Ivan!

I think the problematic sequence is: onSeqNoCompleted(seqNo) → close() → responseHeadersCollector.finish() → responseHeadersCollector.collect() → driverContext.removeAsyncAction(), where responseHeadersCollector.finish() is called before responseHeadersCollector.collect(). This PR prevents that sequence. However, we should not increase the sequence number in onFailure() (call onSeqNoCompleted), as it may cause a NullPointerException later:

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

Line 168 in b504514

Fetched result = buffers.remove(nextCheckpoint);

[elasticsearchmachine]

Hi @ivancea, I've updated the changelog YAML for you.

[ivancea]

(Answering you here @dnhatn, as to have a thread)

This PR prevents that sequence. However, we should not increase the sequence number in onFailure() (call onSeqNoCompleted), as it may cause a NullPointerException later

From what I see, there's already an if (fetched != null) there on line 170, so no possible NPE, unless you mean something else or I missed something 👀

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

Lines 165 to 174 in 9f927b2

	private void discardResults() {
	long nextCheckpoint;
	while ((nextCheckpoint = checkpoint.getPersistedCheckpoint() + 1) <= checkpoint.getProcessedCheckpoint()) {
	Fetched result = buffers.remove(nextCheckpoint);
	checkpoint.markSeqNoAsPersisted(nextCheckpoint);
	if (result != null) {
	releaseFetchedOnAnyThread(result);
	}
	}
	}

[dnhatn]

Thanks Ivan. I took another closer look. I think we are all good.

[dnhatn]

LGTM. Thanks Ivan!

[dnhatn]

Thanks Ivan. I took another closer look. I think we are all good.