[Search] Exceptions in datanodes leading to assertFirstRun() failures

[ivancea]

While debugging CircuitBreakingExceptions in _search with 3 or more nodes, I found that sometimesit throws an assertion error. The first call to assertFirstRun() was:

First exception stacktrace

org.elasticsearch.ElasticsearchException: executed already
	at org.elasticsearch.action.ActionListener$3.assertFirstRun(ActionListener.java:403)
	at org.elasticsearch.action.ActionListener$3.onFailure(ActionListener.java:422)
	at org.elasticsearch.action.search.AbstractSearchAsyncAction.raisePhaseFailure(AbstractSearchAsyncAction.java:647)
	at org.elasticsearch.action.search.AbstractSearchAsyncAction.onPhaseFailure(AbstractSearchAsyncAction.java:624)
	at org.elasticsearch.action.search.SearchQueryThenFetchAsyncAction.onPhaseFailure(SearchQueryThenFetchAsyncAction.java:81)
	at org.elasticsearch.action.search.SearchQueryThenFetchAsyncAction$1.handleException(SearchQueryThenFetchAsyncAction.java:536)
	at org.elasticsearch.transport.TransportService$UnregisterChildTransportResponseHandler.handleException(TransportService.java:1731)
	at org.elasticsearch.transport.TransportService$ContextRestoreResponseHandler.handleException(TransportService.java:1511)
	at org.elasticsearch.transport.InboundHandler.doHandleException(InboundHandler.java:510)
	at org.elasticsearch.transport.InboundHandler.handleException(InboundHandler.java:497)
	at org.elasticsearch.transport.InboundHandler.handlerResponseError(InboundHandler.java:488)
	at org.elasticsearch.transport.InboundHandler.executeResponseHandler(InboundHandler.java:164)
	at org.elasticsearch.transport.InboundHandler.messageReceived(InboundHandler.java:134)
	at org.elasticsearch.transport.InboundHandler.inboundMessage(InboundHandler.java:102)
	at org.elasticsearch.transport.TcpTransport.inboundMessage(TcpTransport.java:830)
	at org.elasticsearch.transport.InboundPipeline.forwardFragment(InboundPipeline.java:116)
	at org.elasticsearch.transport.InboundPipeline.lambda$doHandleBytes$1(InboundPipeline.java:78)
	at org.elasticsearch.transport.InboundDecoder.finishMessage(InboundDecoder.java:144)
	at org.elasticsearch.transport.InboundDecoder.internalDecode(InboundDecoder.java:129)
	at org.elasticsearch.transport.InboundDecoder.decode(InboundDecoder.java:58)
	at org.elasticsearch.transport.InboundPipeline.doHandleBytes(InboundPipeline.java:79)
	at org.elasticsearch.transport.InboundPipeline.handleBytes(InboundPipeline.java:69)
	at org.elasticsearch.transport.netty4.Netty4MessageInboundHandler.channelRead(Netty4MessageInboundHandler.java:54)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at org.elasticsearch.transport.netty4.NettyByteBufSizer.channelRead(NettyByteBufSizer.java:33)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:796)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysPlain(NioEventLoop.java:697)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:660)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at java.base/java.lang.Thread.run(Thread.java:1447)

The second call triggering the assertion had an identical stacktrace.

I reproduced this in main (a59c182, when the issue was opened) with this test:

AggregationsCircuitBreakingIt.java

/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the "Elastic License
 * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
 * Public License v 1"; you may not use this file except in compliance with, at
 * your election, the "Elastic License 2.0", the "GNU Affero General Public
 * License v3.0 only", or the "Server Side Public License, v 1".
 */

package org.elasticsearch.aggregations.bucket;

import com.carrotsearch.randomizedtesting.annotations.Repeat;

import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchPhaseExecutionException;
import org.elasticsearch.aggregations.AggregationIntegTestCase;
import org.elasticsearch.common.breaker.CircuitBreakingException;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.search.aggregations.bucket.composite.TermsValuesSourceBuilder;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.xcontent.XContentBuilder;
import org.elasticsearch.xcontent.XContentFactory;
import org.hamcrest.core.Every;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.IntStream;
import java.util.stream.LongStream;

import static org.elasticsearch.indices.breaker.HierarchyCircuitBreakerService.REQUEST_CIRCUIT_BREAKER_LIMIT_SETTING;
import static org.elasticsearch.indices.breaker.HierarchyCircuitBreakerService.REQUEST_CIRCUIT_BREAKER_TYPE_SETTING;
import static org.elasticsearch.indices.breaker.HierarchyCircuitBreakerService.TOTAL_CIRCUIT_BREAKER_LIMIT_SETTING;
import static org.elasticsearch.indices.breaker.HierarchyCircuitBreakerService.USE_REAL_MEMORY_USAGE_SETTING;
import static org.elasticsearch.search.aggregations.AggregationBuilders.composite;
import static org.elasticsearch.search.aggregations.AggregationBuilders.topHits;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.hamcrest.Matchers.instanceOf;

@ESIntegTestCase.ClusterScope(numDataNodes = 3)
@Repeat(iterations = 10)
public class AggregationsCircuitBreakingIT extends AggregationIntegTestCase {
    @Override
    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
        return Settings.builder()
            .put(super.nodeSettings(nodeOrdinal, otherSettings))
            .put(REQUEST_CIRCUIT_BREAKER_TYPE_SETTING.getKey(), "memory")
            .build();
    }

    private void createIndex() throws IOException {
        XContentBuilder mappingBuilder = XContentFactory.jsonBuilder();
        mappingBuilder.startObject();
        mappingBuilder.startObject("properties");
        {
            mappingBuilder.startObject("integer");
            mappingBuilder.field("type", "integer");
            mappingBuilder.endObject();
        }
        {
            mappingBuilder.startObject("long");
            mappingBuilder.field("type", "long");
            mappingBuilder.endObject();
        }

        mappingBuilder.endObject(); // properties
        mappingBuilder.endObject();

        assertAcked(
            prepareCreate("index").setSettings(Settings.builder().put("index.number_of_shards", randomIntBetween(1, 10)).build())
                .setMapping(mappingBuilder)
        );

        int minimumCombinations = 10_000_000; // To fill the 65k buckets in the composite agg, times the 100 top_hits count
        int docCount = randomIntBetween(100, 10000); // 100 at least to fill the top_hits agg
        int integerFieldMvCount = randomIntBetween(100, (int) Math.ceil((double) minimumCombinations / docCount));
        int longFieldMvCount = (int) Math.ceil((double) minimumCombinations / docCount / integerFieldMvCount);

        List<IndexRequestBuilder> docs = new ArrayList<>();
        for (int i = 0; i < docCount; i++) {
            XContentBuilder docSource = XContentFactory.jsonBuilder();
            docSource.startObject();
            final int docNumber = i;
            List<Integer> integerValues = IntStream.range(0, integerFieldMvCount).map(x -> docNumber + x * 100).boxed().toList();
            List<Long> longValues = LongStream.range(0, longFieldMvCount).map(x -> docNumber + x * 100).boxed().toList();
            docSource.field("integer", integerValues);
            docSource.field("long", longValues);
            docSource.endObject();

            docs.add(prepareIndex("index").setOpType(DocWriteRequest.OpType.CREATE).setSource(docSource));
        }
        indexRandom(true, false, docs);
    }

    public void testCBTrippingOnBigComposite() throws IOException {
        createIndex();

        updateClusterSettings(
            Settings.builder()
                .put(TOTAL_CIRCUIT_BREAKER_LIMIT_SETTING.getKey(), "100MB")
                .put(REQUEST_CIRCUIT_BREAKER_LIMIT_SETTING.getKey(), "10MB")
                .put(USE_REAL_MEMORY_USAGE_SETTING.getKey(), true)
        );

        var searchRequestBuilder = prepareSearch("index").setSize(0)
            .addAggregation(
                composite(
                    "composite",
                    List.of(new TermsValuesSourceBuilder("integer").field("integer"), new TermsValuesSourceBuilder("long").field("long"))
                ).size(65536).subAggregation(
                    topHits("top_hits").size(100)
                )
            );

        try {
            searchRequestBuilder.get();

            fail("Expected the breaker to trip");
        } catch (SearchPhaseExecutionException e) {
            assertThat(List.of(e.guessRootCauses()), Every.everyItem(instanceOf(CircuitBreakingException.class)));
        } finally {
            updateClusterSettings(
                Settings.builder()
                    .putNull(TOTAL_CIRCUIT_BREAKER_LIMIT_SETTING.getKey())
                    .putNull(REQUEST_CIRCUIT_BREAKER_LIMIT_SETTING.getKey())
                    .putNull(USE_REAL_MEMORY_USAGE_SETTING.getKey())
            );
        }
    }
}

[elasticsearchmachine]

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

[cbuescher]

fyi I observed a similar error in a test I modified recently, related PR: #131085
We are still discussing where to fix this but lets link those issues.

[ivancea]

@cbuescher Oh, nice to know! As you're already taking care of this, we can close this one (as you prefer). I'll assign this one to you for now as you're already taking care. Thx!

[cbuescher]

Establishing link to #134151. First of all I was able to reproduce with this seed "-Dtests.seed=E5524FB23B3BF049:709C06F1EF843BB9". Also the test passes in that case when disabling batched query execution with

updateClusterSettings(Settings.builder().put(SearchService.BATCHED_QUERY_PHASE.getKey(), false));
...
updateClusterSettings(Settings.builder().putNull(SearchService.BATCHED_QUERY_PHASE.getKey()));

I'm trying to pin down if the code paths that raise the failures here for the CBEs are similar to the ones we saw in #134151

[cbuescher]

This is a similar situation to what the sorting-test in #134151 shows. The test here provokes CBEs on the data nodes. The coordinating node then receives multiple callbacks from each failing node via SearchQueryThenFetchAsyncAction#handleException that get propagated to onPhaseFailure which triggers the assertion later in the connected ActionListener.

@javanna since you wanted to look into this again more deeply, I currently can only see how we can de-dup multiple failures that happen during reduction on the data nodes either in SearchQueryThenFetchAsyncAction#handleException, higher up like suggested in #134151 or we need to drop the assumption that those listeners can only be called once when we have query batching enabled.

[cbuescher]

Left test repro here

[benchaplin]

Alright... I finally had a chance to dig into this issue. Here's my best attempt at a summarizing how all this works, and how things changed with batched execution (cc: @javanna, @cbuescher).

Some prereqs

Definition: A reduction failure is any exception that may be assigned to the QueryPhaseResultConsumer.failure field. These occur while consuming or reducing shard query results.
For example: A CircuitBreakingException thrown during a partial reduction.

Definition: A phase failure occurs during execution of a search phase and immediately fails the search request, without the possibility of partial results.

Fact: A reduction failure causes a phase failure.

How does a reduction failure lead to a phase failure in non-batched requests?

The query phase executes on every shard. For each result (AbstractSearchAsyncAction#onShardResult):

1. Consume the shard result (QueryPhaseResultConsumer#consumeResult -> consume).
2. Maybe perform a partial reduction (QueryPhaseResultConsumer#tryExecuteNext -> partialReduce).
3. Finish processing the shard result (AbstractSearchAsyncAction#finishOneShard). If it's the last shard, complete the phase.

A reduction failure can occur during either 1 or 2. If it does, it's persisted in QueryPhaseResultConsumer.failure. All subsequent consuming/reducing of shard results is skipped (it's all synchronized and protected by checks against QueryPhaseResultConsumer.failure#get).

Once the last shard is processed, wrap up the phase (AbstractSearchAsyncAction#onPhaseDone). Enter the next phase: the fetch phase.

In the fetch phase, a final reduction occurs (FetchSearchPhase#innerRun -> QueryPhaseResultConsumer#reduce). The final reduction checks whether a reduction failure has occurred. If so, it throws an exception, and a fetch phase failure is raised.

How does a reduction failure lead to a phase failure in batched requests?

Coordinating node: Shards to query are batched per node.

Data node: The query phase executes on each shard in the batch (SearchQueryThenFetchAsyncAction#executeShardTasks). For each result:

1. Same as 1 above (QueryPerNodeState#consumeResult -> QueryPhaseResultConsumer#consumeResult -> consume).
2. Same as 2 above.
3. Finish processing the shard result (QueryPerNodeState#onShardDone). Unless it's the last shard, do nothing.

Same as above, a reduction failure can occur during either 1 or 2.

Once the last shard is processed, check whether a reduction failure has occurred. If so, fail the batched query request (QueryPerNodeState#handleMergeFailure).

Coordinating node: Handle the failed batched query request (In SearchQueryThenFetchAsyncAction#doRun, TransportResponseHandler#handleException). If the error is a reduction failure, a query phase failure is raised.

Summary

The presenting issue here occurs when multiple batched requests experience reduction failures. Multiple threads may then fail the query phase, however the call to listener.onFailure in AbstractSearchAsyncAction#raisePhaseFailure can only be made once.

I see two ways forward:

1. In the style of @cbuescher's Fix assertion error on multiple node search reduce errors #131085, use a flag to ensure the phase failure call is only made once. I'd argue to adjust this approach a bit, raising up the flag into SearchQueryThenFetchAsyncAction#doRun, preventing onPhaseFailure from being called twice there.
2. Engineer the phase failure to occur during final reduction in the fetch phase, as it works for non-batched.

The argument for 1: it fails faster (the phase failure occurs during the query phase as soon as a batched request comes back with a reduction failure).
The argument for 2: it fails more traditionally (before batched, we waited for all shards to come back and didn't fail until final reduction in the fetch phase).

I can't think of any reasons to "fail more traditionally," so I prefer 1. But maybe I'm missing something.

[cbuescher]

@benchaplin what an amazing analysis and write-up, I really enjoyed reading this and will recap the different steps again with a closer look at the code for sure. I'll also take a look at adjusting #131085 later in the way you propose and will ping you there.