Fix/cleanup two spots in open-PIT request handling (#116553) (#117430)

Two things fixed here:
1. Don't fork just to send the response, it's unnecessary. Serializing
   the ID might take a little time but if it's really an issue we should
   optimize it rather than forking just to send a single response.
2. Handle finding a connection cleanly, don't allow the exception to
   bubble up and fail the phase, this may cause leaks.

Author: original-brownbear

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

@@ -28,7 +28,6 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.shard.ShardId;
@@ -257,16 +256,17 @@ protected void executePhaseOnShard(
                     SearchShardTarget shard,
                     SearchActionListener<SearchPhaseResult> phaseListener
                 ) {
-                    final ShardOpenReaderRequest shardRequest = new ShardOpenReaderRequest(
-                        shardIt.shardId(),
-                        shardIt.getOriginalIndices(),
-                        pitRequest.keepAlive()
-                    );
-                    Transport.Connection connection = connectionLookup.apply(shardIt.getClusterAlias(), shard.getNodeId());
+                    final Transport.Connection connection;
+                    try {
+                        connection = connectionLookup.apply(shardIt.getClusterAlias(), shard.getNodeId());
+                    } catch (Exception e) {
+                        phaseListener.onFailure(e);
+                        return;
+                    }
                     transportService.sendChildRequest(
                         connection,
                         OPEN_SHARD_READER_CONTEXT_NAME,
-                        shardRequest,
+                        new ShardOpenReaderRequest(shardIt.shardId(), shardIt.getOriginalIndices(), pitRequest.keepAlive()),
                         task,
                         new ActionListenerResponseHandler<>(
                             phaseListener,
@@ -279,29 +279,9 @@ protected void executePhaseOnShard(
                 @Override
                 protected SearchPhase getNextPhase() {
                     return new SearchPhase(getName()) {
-
-                        private void onExecuteFailure(Exception e) {
-                            onPhaseFailure(this, "sending response failed", e);
-                        }
-
                         @Override
                         public void run() {
-                            execute(new AbstractRunnable() {
-                                @Override
-                                public void onFailure(Exception e) {
-                                    onExecuteFailure(e);
-                                }
-
-                                @Override
-                                protected void doRun() {
-                                    sendSearchResponse(SearchResponseSections.EMPTY_WITH_TOTAL_HITS, results.getAtomicArray());
-                                }
-
-                                @Override
-                                public boolean isForceExecution() {
-                                    return true; // we already created the PIT, no sense in rejecting the task that sends the response.
-                                }
-                            });
+                            sendSearchResponse(SearchResponseSections.EMPTY_WITH_TOTAL_HITS, results.getAtomicArray());
                         }
                     };
                 }