Prevent Leaking Search Tasks on Exceptions in FetchSearchPhase and DfsQueryPhase

[original-brownbear]

• If counter.onResult throws an exception we might leak a transport task because the failure is not handled as a phase failure (instead it bubbles up in the transport service eventually hitting the onFailure callback again and couting down the counter twice).

[elasticmachine]

Pinging @elastic/es-search

[jimczi]

This failure is not fatal so it can stay at this level I think. The bug that you found is that we call this handler when a phase exception happens (finishPhase.run throw an exception during executeFetch) so this leaves a zombie task/connection in the node and the response is never sent back. Changing the FetchSearchPhase with:

diff --git a/server/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java b/server/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java
index 2115b4fa998..548c21ab1b6 100644
--- a/server/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java
+++ b/server/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java
@@ -163,7 +163,11 @@ final class FetchSearchPhase extends SearchPhase {
             new SearchActionListener<FetchSearchResult>(shardTarget, shardIndex) {
                 @Override
                 public void innerOnResponse(FetchSearchResult result) {
-                    counter.onResult(result);
+                    try {
+                        counter.onResult(result);
+                    } catch (Exception e) {
+                        context.onPhaseFailure(FetchSearchPhase.this, "", e);
+                    }
                 }
                 @Override

should be enough and in this case the error will be properly propagated to the user.

[original-brownbear]

@jimczi

I applied the above patch now (to the fetch phase as well as the dfs query phase which has the same issue as far as I can see).
However, I could not for the life of me find a way to trigger this kind of failure in an integration test (and I'm not sure a unit test with some hand crafted failure in org.elasticsearch.action.search.FetchSearchPhase#moveToNextPhase would mean anything).
So I figured I'd assert that we simply don't have any exceptions here (as far as I can tell any exception would be a bug in org.elasticsearch.action.search.SearchPhaseController#merge since org.elasticsearch.action.search.AbstractSearchAsyncAction#executeNextPhase catches all the exceptions and passes them to onPhaseFailure as well).

WDYT?

[original-brownbear]

Jenkins run elasticsearch-ci/2

(watcher failure)

[jimczi]

Thanks @original-brownbear . I left one comment regarding the additional assert. Can you also change the title and description to better reflect what this pr is doing ? Regarding the addition of tests, it is difficult to add since the filling of search hits is not supposed to throw exceptions. IMO this is a bug that we can have an exception at this point so I see this pr as a protection against it but we should also have a follow up to ensure that any discrepancy between shards is caught during the merge and not in the final step. I'll take a look when this pr is merged.

[original-brownbear]

Jenkins run elasticsearch-ci/1

[original-brownbear]

Thanks @jimczi , you're right the assert probably doesn't add anything -> removed it in 1547b57 :)

[jimczi]

LGTM, great catch @original-brownbear !

[jimczi]

Since the fix is simple can we backport this to 6.x and 7.3.x too ?

[original-brownbear]

@jimczi sure will back port to 6.8 and 7.3 as well :)