import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
@Override
public Connection openConnection(DiscoveryNode node, ConnectionProfile profile) throws IOException {
    return new Connection() {
        @Override
        public DiscoveryNode getNode() {
            return node;
        }

        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {
            requests.put(requestId, Tuple.tuple(node, action));
            capturedRequests.add(new CapturedRequest(node, requestId, action, request));
        }

        @Override
        public void close() throws IOException {

        }
    };
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
private void sendSetBanRequest(DiscoveryNodes nodes, BanParentTaskRequest request, ActionListener<Void> listener) {
    for (ObjectObjectCursor<String, DiscoveryNode> node : nodes.getNodes()) {
        logger.trace("Sending ban for tasks with the parent [{}] to the node [{}], ban [{}]", request.parentTaskId, node.key,
            request.ban);
        transportService.sendRequest(node.value, BAN_PARENT_ACTION_NAME, request,
            new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
                @Override
                public void handleResponse(TransportResponse.Empty response) {
                    listener.onResponse(null);
                }

                @Override
                public void handleException(TransportException exp) {
                    logger.warn("Cannot send ban for tasks with the parent [{}] to the node [{}]", request.parentTaskId, node.key);
                    listener.onFailure(exp);
                }
            });
    }
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
@Override
public Connection getConnection(DiscoveryNode node) {
    return new Connection() {
        @Override
        public DiscoveryNode getNode() {
            return node;
        }

        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {

        }

        @Override
        public void close() throws IOException {

        }
    };
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
public void testFailureWithoutRetry() throws Exception {
    Request request = new Request().index("test");
    request.shardId = new ShardId("test", "_na_", 0);
    PlainActionFuture<Response> listener = new PlainActionFuture<>();
    setState(clusterService, ClusterStateCreationUtils.state("test", randomBoolean(), ShardRoutingState.STARTED));

    action.new AsyncSingleAction(request, listener).start();
    assertThat(transport.capturedRequests().length, equalTo(1));
    long requestId = transport.capturedRequests()[0].requestId;
    transport.clear();
    // this should not trigger retry or anything and the listener should report exception immediately
    transport.handleRemoteError(requestId, new TransportException("a generic transport exception", new Exception("generic test exception")));

    try {
        // result should return immediately
        assertTrue(listener.isDone());
        listener.get();
        fail("this should fail with a transport exception");
    } catch (ExecutionException t) {
        if (ExceptionsHelper.unwrap(t, TransportException.class) == null) {
            logger.info("expected TransportException  but got ", t);
            fail("expected and TransportException");
        }
    }
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
@Override
public Connection getConnection(DiscoveryNode node) {
    return new FilteredConnection(transport.getConnection(node)) {
        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {
            DelegateTransport.this.sendRequest(connection, requestId, action, request, options);
        }
    };
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
@Override
public Connection openConnection(DiscoveryNode node, ConnectionProfile profile) throws IOException {
    return new FilteredConnection(transport.openConnection(node, profile)) {
        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {
            DelegateTransport.this.sendRequest(connection, requestId, action, request, options);
        }
    };
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
/**
 * returns the number of in flight operations on primary. -1 upon error.
 */
protected void getInflightOpsCount(final ShardId shardId, ClusterState state, IndexShardRoutingTable shardRoutingTable, final ActionListener<InFlightOpsResponse> listener) {
    try {
        final ShardRouting primaryShard = shardRoutingTable.primaryShard();
        final DiscoveryNode primaryNode = state.nodes().get(primaryShard.currentNodeId());
        if (primaryNode == null) {
            logger.trace("{} failed to resolve node for primary shard {}, skipping sync", shardId, primaryShard);
            listener.onResponse(new InFlightOpsResponse(-1));
            return;
        }
        logger.trace("{} retrieving in flight operation count", shardId);
        transportService.sendRequest(primaryNode, IN_FLIGHT_OPS_ACTION_NAME, new InFlightOpsRequest(shardId),
                new TransportResponseHandler<InFlightOpsResponse>() {
                    @Override
                    public InFlightOpsResponse newInstance() {
                        return new InFlightOpsResponse();
                    }

                    @Override
                    public void handleResponse(InFlightOpsResponse response) {
                        listener.onResponse(response);
                    }

                    @Override
                    public void handleException(TransportException exp) {
                        logger.debug("{} unexpected error while retrieving in flight op count", shardId);
                        listener.onFailure(exp);
                    }

                    @Override
                    public String executor() {
                        return ThreadPool.Names.SAME;
                    }
                });
    } catch (Exception e) {
        listener.onFailure(e);
    }
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
@Override
public void handleException(TransportException exp) {
    logger.debug((Supplier<?>) () -> new ParameterizedMessage("shards active request failed for {}", shardId), exp);
    if (awaitingResponses.decrementAndGet() == 0) {
        allNodesResponded();
    }
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
private void sendCommitToNode(final DiscoveryNode node, final ClusterState clusterState, final SendingController sendingController) {
    try {
        logger.trace("sending commit for cluster state (uuid: [{}], version [{}]) to [{}]",
            clusterState.stateUUID(), clusterState.version(), node);
        TransportRequestOptions options = TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STATE).build();
        // no need to put a timeout on the options here, because we want the response to eventually be received
        // and not log an error if it arrives after the timeout
        transportService.sendRequest(node, COMMIT_ACTION_NAME,
                new CommitClusterStateRequest(clusterState.stateUUID()),
                options,
                new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {

                    @Override
                    public void handleResponse(TransportResponse.Empty response) {
                        if (sendingController.getPublishingTimedOut()) {
                            logger.debug("node {} responded to cluster state commit [{}]", node, clusterState.version());
                        }
                        sendingController.getPublishResponseHandler().onResponse(node);
                    }

                    @Override
                    public void handleException(TransportException exp) {
                        logger.debug((org.apache.logging.log4j.util.Supplier<?>) () ->
                            new ParameterizedMessage("failed to commit cluster state (uuid [{}], version [{}]) to {}",
                                clusterState.stateUUID(), clusterState.version(), node), exp);
                        sendingController.getPublishResponseHandler().onFailure(node, exp);
                    }
                });
    } catch (Exception t) {
        logger.warn((org.apache.logging.log4j.util.Supplier<?>) () ->
            new ParameterizedMessage("error sending cluster state commit (uuid [{}], version [{}]) to {}",
                clusterState.stateUUID(), clusterState.version(), node), t);
        sendingController.getPublishResponseHandler().onFailure(node, t);
    }
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
protected TransportResponseHandler<UnicastPingResponse> getPingResponseHandler(final PingingRound pingingRound,
                                                                               final DiscoveryNode node) {
    return new TransportResponseHandler<UnicastPingResponse>() {

        @Override
        public UnicastPingResponse newInstance() {
            return new UnicastPingResponse();
        }

        @Override
        public String executor() {
            return ThreadPool.Names.SAME;
        }

        @Override
        public void handleResponse(UnicastPingResponse response) {
            logger.trace("[{}] received response from {}: {}", pingingRound.id(), node, Arrays.toString(response.pingResponses));
            if (pingingRound.isClosed()) {
                if (logger.isTraceEnabled()) {
                    logger.trace("[{}] skipping received response from {}. already closed", pingingRound.id(), node);
                }
            } else {
                Stream.of(response.pingResponses).forEach(pingingRound::addPingResponseToCollection);
            }
        }

        @Override
        public void handleException(TransportException exp) {
            if (exp instanceof ConnectTransportException || exp.getCause() instanceof ConnectTransportException) {
                // ok, not connected...
                logger.trace((Supplier<?>) () -> new ParameterizedMessage("failed to connect to {}", node), exp);
            } else if (closed == false) {
                logger.warn((Supplier<?>) () -> new ParameterizedMessage("failed to send ping to [{}]", node), exp);
            }
        }
    };
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
public void allocateDangled(Collection<IndexMetaData> indices, final Listener listener) {
    ClusterState clusterState = clusterService.state();
    DiscoveryNode masterNode = clusterState.nodes().getMasterNode();
    if (masterNode == null) {
        listener.onFailure(new MasterNotDiscoveredException("no master to send allocate dangled request"));
        return;
    }
    AllocateDangledRequest request = new AllocateDangledRequest(clusterService.localNode(), indices.toArray(new IndexMetaData[indices.size()]));
    transportService.sendRequest(masterNode, ACTION_NAME, request, new TransportResponseHandler<AllocateDangledResponse>() {
        @Override
        public AllocateDangledResponse newInstance() {
            return new AllocateDangledResponse();
        }

        @Override
        public void handleResponse(AllocateDangledResponse response) {
            listener.onResponse(response);
        }

        @Override
        public void handleException(TransportException exp) {
            listener.onFailure(exp);
        }

        @Override
        public String executor() {
            return ThreadPool.Names.SAME;
        }
    });
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
private void sendShardAction(final String actionName, final ClusterState currentState, final ShardEntry shardEntry, final Listener listener) {
    ClusterStateObserver observer = new ClusterStateObserver(currentState, clusterService, null, logger, threadPool.getThreadContext());
    DiscoveryNode masterNode = currentState.nodes().getMasterNode();
    Predicate<ClusterState> changePredicate = MasterNodeChangePredicate.build(currentState);
    if (masterNode == null) {
        logger.warn("{} no master known for action [{}] for shard entry [{}]", shardEntry.shardId, actionName, shardEntry);
        waitForNewMasterAndRetry(actionName, observer, shardEntry, listener, changePredicate);
    } else {
        logger.debug("{} sending [{}] to [{}] for shard entry [{}]", shardEntry.shardId, actionName, masterNode.getId(), shardEntry);
        transportService.sendRequest(masterNode,
            actionName, shardEntry, new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
                @Override
                public void handleResponse(TransportResponse.Empty response) {
                    listener.onSuccess();
                }

                @Override
                public void handleException(TransportException exp) {
                    if (isMasterChannelException(exp)) {
                        waitForNewMasterAndRetry(actionName, observer, shardEntry, listener, changePredicate);
                    } else {
                        logger.warn((Supplier<?>) () -> new ParameterizedMessage("{} unexpected failure while sending request [{}] to [{}] for shard entry [{}]", shardEntry.shardId, actionName, masterNode, shardEntry), exp);
                        listener.onFailure(exp instanceof RemoteTransportException ? (Exception) (exp.getCause() instanceof Exception ? exp.getCause() : new ElasticsearchException(exp.getCause())) : exp);
                    }
                }
            });
    }
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
private void fetchShardsInternal(SearchRequest searchRequest, List<String> indices,
                                 final ActionListener<ClusterSearchShardsResponse> listener) {
    final DiscoveryNode node = nodeSupplier.get();
    ClusterSearchShardsRequest searchShardsRequest = new ClusterSearchShardsRequest(indices.toArray(new String[indices.size()]))
        .indicesOptions(searchRequest.indicesOptions()).local(true).preference(searchRequest.preference())
        .routing(searchRequest.routing());
    transportService.sendRequest(node, ClusterSearchShardsAction.NAME, searchShardsRequest,
        new TransportResponseHandler<ClusterSearchShardsResponse>() {

            @Override
            public ClusterSearchShardsResponse newInstance() {
                return new ClusterSearchShardsResponse();
            }

            @Override
            public void handleResponse(ClusterSearchShardsResponse clusterSearchShardsResponse) {
                listener.onResponse(clusterSearchShardsResponse);
            }

            @Override
            public void handleException(TransportException e) {
                listener.onFailure(e);
            }

            @Override
            public String executor() {
                return ThreadPool.Names.SEARCH;
            }
        });
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
@Override
public void handleException(TransportException exp) {
    logger.warn((Supplier<?>)
        () -> new ParameterizedMessage("fetching nodes from external cluster {} failed", clusterAlias),
        exp);
    try {
        IOUtils.closeWhileHandlingException(connection);
    } finally {
        // once the connection is closed lets try the next node
        collectRemoteNodes(seedNodes, transportService, listener);
    }
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
private void sendNodeRequest(final DiscoveryNode node, List<ShardRouting> shards, final int nodeIndex) {
    try {
        NodeRequest nodeRequest = new NodeRequest(node.getId(), request, shards);
        if (task != null) {
            nodeRequest.setParentTask(clusterService.localNode().getId(), task.getId());
        }
        transportService.sendRequest(node, transportNodeBroadcastAction, nodeRequest, new TransportResponseHandler<NodeResponse>() {
            @Override
            public NodeResponse newInstance() {
                return new NodeResponse();
            }

            @Override
            public void handleResponse(NodeResponse response) {
                onNodeResponse(node, nodeIndex, response);
            }

            @Override
            public void handleException(TransportException exp) {
                onNodeFailure(node, nodeIndex, exp);
            }

            @Override
            public String executor() {
                return ThreadPool.Names.SAME;
            }
        });
    } catch (Exception e) {
        onNodeFailure(node, nodeIndex, e);
    }
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
public void start() {
    if (shardIt == null) {
        // just execute it on the local node
        transportService.sendRequest(clusterService.localNode(), transportShardAction, internalRequest.request(), new TransportResponseHandler<Response>() {
            @Override
            public Response newInstance() {
                return newResponse();
            }

            @Override
            public String executor() {
                return ThreadPool.Names.SAME;
            }

            @Override
            public void handleResponse(final Response response) {
                listener.onResponse(response);
            }

            @Override
            public void handleException(TransportException exp) {
                listener.onFailure(exp);
            }
        });
    } else {
        perform(null);
    }
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
@Override
protected TransportResponseHandler<UnicastPingResponse> getPingResponseHandler(PingingRound pingingRound, DiscoveryNode node) {
    markTaskAsStarted("ping [" + node + "]");
    TransportResponseHandler<UnicastPingResponse> original = super.getPingResponseHandler(pingingRound, node);
    return new TransportResponseHandler<UnicastPingResponse>() {
        @Override
        public UnicastPingResponse newInstance() {
            return original.newInstance();
        }

        @Override
        public void handleResponse(UnicastPingResponse response) {
            original.handleResponse(response);
            markTaskAsCompleted("ping [" + node + "]");
        }

        @Override
        public void handleException(TransportException exp) {
            original.handleException(exp);
            markTaskAsCompleted("ping [" + node + "] (error)");
        }

        @Override
        public String executor() {
            return original.executor();
        }
    };
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
@Override
public AsyncSender interceptSender(AsyncSender sender) {
    return new AsyncSender() {
        @Override
        public <T extends TransportResponse> void sendRequest(Transport.Connection connection, String action,
                                                              TransportRequest request,
                                                              TransportRequestOptions options,
                                                              TransportResponseHandler<T> handler) {
            final ClusterName clusterName = new ClusterName("cluster1");
            if (TransportLivenessAction.NAME.equals(action)) {
                assertHeaders(threadPool);
                ((TransportResponseHandler<LivenessResponse>) handler).handleResponse(
                    new LivenessResponse(clusterName, connection.getNode()));
            } else if (ClusterStateAction.NAME.equals(action)) {
                assertHeaders(threadPool);
                ClusterName cluster1 = clusterName;
                ClusterState.Builder builder = ClusterState.builder(cluster1);
                //the sniffer detects only data nodes
                builder.nodes(DiscoveryNodes.builder().add(new DiscoveryNode("node_id", "someId", "some_ephemeralId_id",
                    address.address().getHostString(), address.getAddress(), address, Collections.emptyMap(),
                        Collections.singleton(DiscoveryNode.Role.DATA), Version.CURRENT)));
                ((TransportResponseHandler<ClusterStateResponse>) handler)
                        .handleResponse(new ClusterStateResponse(cluster1, builder.build(), 0L));
                clusterStateLatch.countDown();
            } else if (TransportService.HANDSHAKE_ACTION_NAME .equals(action)) {
                ((TransportResponseHandler<TransportService.HandshakeResponse>) handler).handleResponse(
                    new TransportService.HandshakeResponse(connection.getNode(), clusterName, connection.getNode().getVersion()));
            } else {
                handler.handleException(new TransportException("", new InternalException(action)));
            }
        }
    };
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
private <T extends TransportResponse> TransportResponseHandler wrapLivenessResponseHandler(TransportResponseHandler<T> handler,
                                                                                           DiscoveryNode node,
                                                                                           ClusterName clusterName) {
    return new TransportResponseHandler<T>() {
        @Override
        public T newInstance() {
            return handler.newInstance();
        }

        @Override
        @SuppressWarnings("unchecked")
        public void handleResponse(T response) {
            LivenessResponse livenessResponse = new LivenessResponse(clusterName,
                    new DiscoveryNode(node.getName(), node.getId(), node.getEphemeralId(), "liveness-hostname" + node.getId(),
                            "liveness-hostaddress" + node.getId(),
                            livenessAddress, node.getAttributes(), node.getRoles(),
                            node.getVersion()));
            handler.handleResponse((T)livenessResponse);
        }

        @Override
        public void handleException(TransportException exp) {
            handler.handleException(exp);
        }

        @Override
        public String executor() {
            return handler.executor();
        }
    };
}
 

import org.elasticsearch.transport.TransportException; //导入依赖的package包/类
public void publishClusterStateVersionToNode(final long newVersion,
        final DiscoveryNode node, 
        final TimeValue publishTimeout,
        final BlockingClusterStatePublishResponseHandler publishResponseHandler) {
    try {
        BytesStreamOutput bStream = new BytesStreamOutput();
        bStream.writeLong(newVersion);
        clusterService.localNode().writeTo(bStream);
        bStream.writeString(node.getId());
        BytesReference nodeBytes = bStream.bytes();
        TransportRequestOptions options = TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STATE).withCompress(false).withTimeout(publishTimeout).build();
        transportService.sendRequest(node, PUBLISH_VERSION_ACTION_NAME, new BytesTransportRequest(nodeBytes, node.version()), 
                options, 
                new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
            
            @Override
            public void handleResponse(TransportResponse.Empty response) {
                publishResponseHandler.onResponse(node);
            }
            
            @Override
            public void handleException(TransportException exp) {
                logger.debug("failed to send cluster state to {}, version {}", exp, node, newVersion);
                publishResponseHandler.onFailure(node, exp);
            }
        });
    } catch (Throwable t) {
        logger.warn("error sending cluster state to {}", t, node);
        publishResponseHandler.onFailure(node, t);
    }
}