Remove early phase failure in batched

[benchaplin]

Resolves #134151, #130821.

Background

A bug was introduced by #121885 due to the following code, which handles batched query exceptions due to a batched partial reduction failure:

elasticsearch/server/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java

Lines 525 to 544 in bd35649

	@Override
	public void handleException(TransportException e) {
	Exception cause = (Exception) ExceptionsHelper.unwrapCause(e);
	logger.debug("handling node search exception coming from [" + nodeId + "]", cause);
	if (e instanceof SendRequestTransportException || cause instanceof TaskCancelledException) {
	// two possible special cases here where we do not want to fail the phase:
	// failure to send out the request -> handle things the same way a shard would fail with unbatched execution
	// as this could be a transient failure and partial results we may have are still valid
	// cancellation of the whole batched request on the remote -> maybe we timed out or so, partial results may
	// still be valid
	onNodeQueryFailure(e, request, routing);
	} else {
	// Remote failure that wasn't due to networking or cancellation means that the data node was unable to reduce
	// its local results. Failure to reduce always fails the phase without exception so we fail the phase here.
	if (results instanceof QueryPhaseResultConsumer queryPhaseResultConsumer) {
	queryPhaseResultConsumer.failure.compareAndSet(null, cause);
	}
	onPhaseFailure(getName(), "", cause);
	}
	}

Raising a phase failure in this way leads to a couple issues:

1. It can be called more than once (as seen in [Search] Exceptions in datanodes leading to assertFirstRun() failures #134151).
2. The subsequent freeing of contexts can miss concurrent in-flight queries, resulting in open contexts after the failure (as seen in [CI] SearchWithRejectionsIT testOpenContextsAfterRejections failing #130821).

Solution

Problem 1 could be resolved with a simple flag, as proposed in #131085. Problem 2 could be resolved with some careful use of the same flag to clean contexts upon receiving stale query results. However, in the interest of stability, I propose a solution that more closely resembles how a reduction failure is handled by a non-batched query phase. In non-batched, a reduction failure is held in the QueryPhaseResultConsumer until shard fanout is complete. Only later, during final reduction at the beginning of the fetch phase, do we fail the search.

Fast failure + proper task cancellation are worthy goals for the future. I am tracking these as follow-up improvements for after the release of batched query execution.

This PR:

1. Alters a batched query request to respond with shard results in the case of a reduction failure on the data node (the failure is now conditionally included in the NodeQueryResponse).
2. Removes the early phase failure on the coord node. The coord's QueryPhaseResultConsumer will hold onto the failure and fail eventually during the fetch phase, same as non-batched.

[elasticsearchmachine]

Pinging @elastic/es-search-foundations (Team:Search Foundations)

[elasticsearchmachine]

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

[chrisparrinello]

Since we're changing the shape of this message, do we need to create a new transport version or is that taken care of for us?

[benchaplin]

Yes I believe I do, once I learn how 😂

[chrisparrinello]

LGTM

[javanna]

I left a couple of minor comments. LGTM otherwise

[javanna]

Is it necessary to refactor this serialization code? Moving it around makes it more difficult to eye ball it somehow.

[javanna]

disregard this previous comment. I understand why you did things the way you did. It's fine as-is. As for the review, I simply trust that the successful writing is a plain copy of the previous code we had, with no changes.

[benchaplin]

That's right, same as the previous code except for the additional out.writeBoolean(false);, telling us there's no reduction failure.

[javanna]

should this be in a finally block ?

[benchaplin]

In the case of an IOException, the caller releases contexts in a catch block. So I think this is alright, else we'll be releasing twice?

[javanna]

Maybe add a comment about where the corresponding read code for this can be found? future readers may be looking for it and not easily finding it, due to the fact that we write to an opaque bytes transport response.

[benchaplin]

Good idea, done.

[javanna]

I wonder if I am missing something: is this doing the same that handleMergeFailure was previously doing?

[benchaplin]

Yes, the exact same. handleMergeFailure doesn't make sense to keep in the new serialization scheme, so I decided to remove the method.

[benchaplin]

A failure might occur during the final data node reduction. In this case, due to the central change of this PR, we still send back QuerySearchResults to sit on the coord node. Therefore we can no longer assert that a QuerySearchResult has been reduced when serializing it.

[elasticsearchmachine]

💔 Backport failed

Status	Branch	Result
❌	9.1	Commit could not be cherrypicked due to conflicts
The backport operation could not be completed due to the following error:

An unhandled error occurred. Please consult the logs

You can use sqren/backport to manually backport by running backport --upstream elastic/elasticsearch --pr 136889