flowchart TB
    subgraph "上层模块"
        direction LR
        SearchAction[SearchTransportService<br/>搜索请求]
        ReplicationAction[TransportReplicationAction<br/>复制请求]
        ClusterAction[PublicationTransportHandler<br/>集群状态发布]
    end

    subgraph "传输服务层 TransportService"
        direction TB
        TS[TransportService<br/>传输服务门面]
        subgraph "连接管理"
            CM[ClusterConnectionManager<br/>连接管理器]
            ConnPool[连接池<br/>connectedNodes]
            ConnValidator[ConnectionValidator<br/>连接验证器]
        end
        subgraph "请求处理"
            RH[RequestHandlers<br/>请求处理器注册表]
            ResponseHandlers[ResponseHandlers<br/>响应处理器映射]
            TaskMgr[TaskManager<br/>任务管理器]
        end
        subgraph "拦截与监控"
            Interceptor[TransportInterceptor<br/>传输拦截器]
            Listener[TransportMessageListener<br/>消息监听器]
        end
    end

    subgraph "传输抽象层 Transport"
        direction TB
        Transport[Transport接口<br/>传输层抽象]
        Connection[Connection接口<br/>连接抽象]
        LocalConn[LocalNodeConnection<br/>本地连接优化]
    end

    subgraph "TCP传输实现层 TcpTransport"
        direction TB
        TcpTransport[TcpTransport<br/>TCP传输实现]
        subgraph "出站路径"
            OutboundHandler[OutboundHandler<br/>出站处理器]
            Serializer[序列化器<br/>serialize]
            Compressor[压缩器<br/>compress]
        end
        subgraph "入站路径"
            InboundHandler[InboundHandler<br/>入站处理器]
            Deserializer[反序列化器<br/>deserialize]
            Decompressor[解压缩器<br/>decompress]
        end
        subgraph "连接实现"
            NodeChannels[NodeChannels<br/>节点通道组]
            TcpChannel[TcpChannel<br/>TCP通道]
            ChannelStats[ChannelStats<br/>通道统计]
        end
    end

    subgraph "网络层 Netty"
        direction LR
        NettyTransport[Netty4Transport<br/>Netty实现]
        NioEventLoop[NioEventLoopGroup<br/>事件循环组]
        NettyChannel[Netty Channel<br/>网络通道]
        ByteBuf[ByteBuf<br/>零拷贝缓冲]
    end

    SearchAction --> TS
    ReplicationAction --> TS
    ClusterAction --> TS

    TS --> CM
    TS --> RH
    TS --> ResponseHandlers
    TS --> Interceptor
    TS --> Listener
    TS --> TaskMgr

    CM --> ConnPool
    CM --> ConnValidator
    CM --> Transport

    TS --> Connection
    Connection --> LocalConn
    Connection --> Transport

    Transport --> TcpTransport

    TcpTransport --> OutboundHandler
    TcpTransport --> InboundHandler
    TcpTransport --> NodeChannels

    OutboundHandler --> Serializer
    OutboundHandler --> Compressor
    OutboundHandler --> TcpChannel

    InboundHandler --> Deserializer
    InboundHandler --> Decompressor
    InboundHandler --> RH

    NodeChannels --> TcpChannel
    TcpChannel --> ChannelStats
    TcpChannel --> NettyTransport

    NettyTransport --> NioEventLoop
    NettyTransport --> NettyChannel
    NettyChannel --> ByteBuf

public class TransportService extends AbstractLifecycleComponent {
    protected final Transport transport;
    protected final ConnectionManager connectionManager;
    protected final ThreadPool threadPool;
    private final TaskManager taskManager;
    private final TransportInterceptor interceptor;

    // 发送请求
    public <T extends TransportResponse> void sendRequest(
        DiscoveryNode node,
        String action,
        TransportRequest request,
        TransportResponseHandler<T> handler
    ) {
        // 1. 获取连接
        Transport.Connection connection = getConnection(node);

        // 2. 通过拦截器发送
        asyncSender.sendRequest(connection, action, request, options, handler);
    }

    // 注册请求处理器
    public <Request extends TransportRequest> void registerRequestHandler(
        String action,
        Executor executor,
        Writeable.Reader<Request> requestReader,
        TransportRequestHandler<Request> handler
    ) {
        RequestHandlerRegistry<Request> reg = new RequestHandlerRegistry<>(
            action,
            requestReader,
            taskManager,
            handler,
            executor
        );
        transport.registerRequestHandler(reg);
    }

    // 连接到节点
    public void connectToNode(
        DiscoveryNode node,
        ConnectionProfile connectionProfile,
        ActionListener<Releasable> listener
    ) {
        connectionManager.connectToNode(node, connectionProfile, connectionValidator, listener);
    }
}

// 如果目标是本地节点,直接调用处理器,不走网络
if (isLocalNode(node)) {
    return localNodeConnection;
}

public interface ConnectionManager extends Closeable {
    // 连接到节点
    void connectToNode(
        DiscoveryNode node,
        ConnectionProfile connectionProfile,
        ConnectionValidator validator,
        ActionListener<Releasable> listener
    );

    // 获取连接
    Transport.Connection getConnection(DiscoveryNode node);

    // 断开连接
    void disconnectFromNode(DiscoveryNode node);

    // 获取所有已连接节点
    Set<DiscoveryNode> getAllConnectedNodes();
}

public class ConnectionProfile {
    private int numConnectionsPerType;  // 每种类型的连接数
    private TimeValue connectTimeout;   // 连接超时
    private TimeValue handshakeTimeout; // 握手超时
    private boolean compressionEnabled; // 是否启用压缩

    // 连接类型
    public enum ConnectionTypeHandle {
        RECOVERY,  // 恢复连接
        BULK,      // 批量操作连接
        REG,       // 常规连接
        STATE,     // 集群状态连接
        PING       // Ping 连接
    }
}

public interface Transport extends LifecycleComponent {
    // 发送请求
    void sendRequest(
        Connection connection,
        long requestId,
        String action,
        TransportRequest request,
        TransportRequestOptions options
    );

    // 打开连接
    void openConnection(
        DiscoveryNode node,
        ConnectionProfile profile,
        ActionListener<Connection> listener
    );

    // 注册请求处理器
    <Request extends TransportRequest> void registerRequestHandler(
        RequestHandlerRegistry<Request> reg
    );

    // 响应处理器
    interface ResponseHandlers {
        TransportResponseHandler<?> onResponseReceived(
            long requestId,
            TransportMessageListener listener
        );
        void onResponseSent(long requestId);
    }
}

public abstract class TcpTransport extends AbstractLifecycleComponent implements Transport {
    private final OutboundHandler outboundHandler;
    private final InboundHandler inboundHandler;
    private final ResponseHandlers responseHandlers;

    // 发送请求
    @Override
    public void sendRequest(
        Connection connection,
        long requestId,
        String action,
        TransportRequest request,
        TransportRequestOptions options
    ) throws TransportException {
        // 1. 序列化请求
        BytesReference message = serialize(requestId, action, request);

        // 2. 通过出站处理器发送
        outboundHandler.sendMessage(connection.getChannel(), message, listener);
    }

    // 处理入站数据
    protected void handleInboundMessage(
        TcpChannel channel,
        InboundMessage message
    ) {
        if (message.isRequest()) {
            inboundHandler.handleRequest(channel, message);
        } else {
            inboundHandler.handleResponse(channel, message);
        }
    }
}

public class InboundHandler {
    // 处理请求
    public void handleRequest(TcpChannel channel, InboundMessage message) {
        // 1. 反序列化请求
        TransportRequest request = deserialize(message);

        // 2. 查找处理器
        RequestHandlerRegistry handler = requestHandlers.get(message.getAction());

        // 3. 执行处理器
        executor.execute(() -> {
            try {
                handler.processRequest(request, channel);
            } catch (Exception e) {
                sendErrorResponse(channel, message.getRequestId(), e);
            }
        });
    }

    // 处理响应
    public void handleResponse(TcpChannel channel, InboundMessage message) {
        // 1. 查找响应处理器
        TransportResponseHandler handler = responseHandlers.onResponseReceived(
            message.getRequestId()
        );

        // 2. 反序列化响应
        TransportResponse response = deserialize(message, handler.reader());

        // 3. 回调处理器
        handler.handleResponse(response);
    }
}

public class OutboundHandler {
    // 发送消息
    public void sendMessage(
        TcpChannel channel,
        BytesReference message,
        ActionListener<Void> listener
    ) {
        // 1. 压缩（如果启用）
        if (compressionEnabled && message.length() > compressionThreshold) {
            message = compress(message);
        }

        // 2. 写入通道
        channel.sendMessage(message, listener);
    }
}

classDiagram
    class TransportRequest {
        <<abstract>>
        -TaskId parentTaskId
        +setParentTask(TaskId)
        +getParentTask()
    }

    class TransportResponse {
        <<abstract>>
        +writeTo(StreamOutput)
    }

    class TransportMessage {
        <<interface>>
        +writeTo(StreamOutput)
    }

    TransportMessage <|.. TransportRequest
    TransportMessage <|.. TransportResponse

classDiagram
    class Connection {
        <<interface>>
        +getNode() DiscoveryNode
        +sendRequest(long, String, TransportRequest, TransportRequestOptions)
        +close()
    }

    class ConnectionProfile {
        -Map~ConnectionTypeHandle, ConnectionTypeSettings~ connectionTypeSettings
        -TimeValue connectTimeout
        -TimeValue handshakeTimeout
        -boolean compressionEnabled
        +getNumConnections()
        +getConnectTimeout()
    }

    class ConnectionTypeHandle {
        <<enumeration>>
        RECOVERY
        BULK
        REG
        STATE
        PING
    }

    ConnectionProfile --> ConnectionTypeHandle

classDiagram
    class TransportResponseHandler {
        <<interface>>
        +read(StreamInput) T
        +handleResponse(T)
        +handleException(TransportException)
        +executor() String
    }

    class ActionListenerResponseHandler {
        -ActionListener~T~ listener
        -Writeable.Reader~T~ reader
        -String executor
        +handleResponse(T)
        +handleException(TransportException)
    }

    TransportResponseHandler <|.. ActionListenerResponseHandler

+------------------+
| Header (variable)|  消息头
+------------------+
| Payload (variable)| 消息体
+------------------+

+--------+--------+--------+--------+
| Length | Status | Version| Action |
| 4 bytes| 1 byte | 1 byte | variable |
+--------+--------+--------+--------+
| RequestId (8 bytes)                |
+--------+--------+--------+--------+
| Features (4 bytes)                 |
+--------+--------+--------+--------+

sequenceDiagram
    autonumber
    participant Node1 as Node 1
    participant Node2 as Node 2

    Note over Node1,Node2: 建立 TCP 连接

    Node1->>Node2: TCP SYN
    Node2-->>Node1: TCP SYN-ACK
    Node1->>Node2: TCP ACK

    Note over Node1,Node2: Handshake (握手)

    Node1->>Node2: HandshakeRequest<br/>(version, cluster_name, node_id)
    Node2->>Node2: 验证:<br/>1. 集群名称匹配<br/>2. 版本兼容<br/>3. 许可证兼容

    alt 验证通过
        Node2-->>Node1: HandshakeResponse<br/>(version, node_info)
        Note over Node1,Node2: 连接建立成功
    else 验证失败
        Node2-->>Node1: HandshakeResponse<br/>(error)
        Node2->>Node2: 关闭连接
        Note over Node1,Node2: 连接失败
    end

public <T extends TransportResponse> void sendRequest(
    DiscoveryNode node,
    String action,
    TransportRequest request,
    TransportResponseHandler<T> handler
)

// 发送搜索请求到另一个节点
transportService.sendRequest(
    targetNode,
    "indices:data/read/search",
    searchRequest,
    new TransportResponseHandler<SearchResponse>() {
        @Override
        public SearchResponse read(StreamInput in) throws IOException {
            return new SearchResponse(in);
        }

        @Override
        public void handleResponse(SearchResponse response) {
            // 处理响应
        }

        @Override
        public void handleException(TransportException exp) {
            // 处理异常
        }

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

public <Request extends TransportRequest> void registerRequestHandler(
    String action,
    Executor executor,
    Writeable.Reader<Request> requestReader,
    TransportRequestHandler<Request> handler
)

// 注册搜索请求处理器
transportService.registerRequestHandler(
    "indices:data/read/search",
    ThreadPool.Names.SEARCH,
    SearchRequest::new,
    (request, channel, task) -> {
        // 处理搜索请求
        SearchResponse response = searchService.executeQueryPhase(request, (SearchTask) task);
        channel.sendResponse(response);
    }
);

flowchart TD
    A[TransportSearchAction] --> B[SearchTransportService]
    B --> C[TransportService.sendChildRequest]
    C --> D[AsyncSender.sendRequest]
    D --> E[ConnectionManager.getConnection]
    E --> F[Transport.Connection]
    F --> G[TcpTransport.sendRequest]
    G --> H[OutboundHandler.sendRequest]
    H --> I[OutboundHandler.serialize]
    I --> J[TcpChannel.sendMessage]
    J --> K[Netty Channel.write]

    L[Netty Channel.read] --> M[InboundHandler.inboundMessage]
    M --> N[InboundHandler.messageReceived]
    N --> O[InboundHandler.handleRequest]
    O --> P[RequestHandlerRegistry.processMessageReceived]
    P --> Q[SearchService.executeQueryPhase]
    Q --> R[构建 QuerySearchResult]
    R --> S[TransportChannel.sendResponse]
    S --> T[OutboundHandler.sendResponse]

// server/src/main/java/org/elasticsearch/action/search/SearchTransportService.java
public void sendExecuteQuery(
    Transport.Connection connection,
    final ShardSearchRequest request,
    SearchTask task,
    final ActionListener<SearchPhaseResult> listener
) {
    // 1. 确定是否需要优化（单分片场景）
    final boolean fetchDocuments = request.numberOfShards() == 1
        && (request.source() == null || request.source().rankBuilder() == null);

    // 2. 选择响应读取器（QuerySearchResult 或 QueryFetchSearchResult）
    Writeable.Reader<SearchPhaseResult> reader = fetchDocuments
        ? QueryFetchSearchResult::new
        : in -> new QuerySearchResult(in, true);

    // 3. 通过传输服务发送请求（使用子任务机制）
    transportService.sendChildRequest(
        connection,
        QUERY_ACTION_NAME,  // "indices:data/read/search[phase/query]"
        request,
        task,
        new ConnectionCountingHandler<>(listener, reader, connection)
    );
}

// server/src/main/java/org/elasticsearch/transport/TransportService.java
public <T extends TransportResponse> void sendChildRequest(
    final Transport.Connection connection,
    final String action,
    final TransportRequest request,
    final Task parentTask,
    final TransportResponseHandler<T> handler
) {
    // 1. 设置父任务 ID（用于任务追踪和取消）
    request.setParentTask(parentTask.taskId());

    // 2. 调用标准发送逻辑
    sendRequest(connection, action, request, TransportRequestOptions.EMPTY, handler);
}

public <T extends TransportResponse> void sendRequest(
    final Transport.Connection connection,
    final String action,
    final TransportRequest request,
    final TransportRequestOptions options,
    TransportResponseHandler<T> handler
) {
    // 1. 通过拦截器发送（支持权限控制、监控等）
    asyncSender.sendRequest(connection, action, request, options, handler);
}

// 拦截器处理后的实际发送逻辑
private <T extends TransportResponse> void sendRequestInternal(
    final Transport.Connection connection,
    final String action,
    final TransportRequest request,
    final TransportRequestOptions options,
    TransportResponseHandler<T> handler
) {
    // 1. 生成唯一的请求 ID
    final long requestId = responseHandlers.newRequestId();

    // 2. 设置超时处理器
    final TimeoutHandler timeoutHandler = new TimeoutHandler(requestId, connection.getNode(), action);

    // 3. 注册响应处理器（requestId → handler）
    try {
        responseHandlers.add(requestId, new RequestHolder<>(
            handler,
            connection,
            action,
            timeoutHandler
        ));
    } catch (Exception e) {
        // 注册失败，立即触发异常回调
        handler.handleException(new SendRequestTransportException(connection.getNode(), action, e));
        return;
    }

    // 4. 通过底层连接发送请求
    try {
        connection.sendRequest(requestId, action, request, options);
    } catch (Exception e) {
        // 发送失败，移除响应处理器并触发异常
        responseHandlers.remove(requestId);
        handler.handleException(new SendRequestTransportException(connection.getNode(), action, e));
    }
}

// server/src/main/java/org/elasticsearch/transport/TcpTransport.java (内部类 NodeChannels)
@Override
public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
    throws IOException, TransportException {
    // 1. 检查连接状态
    if (isClosing.get()) {
        throw new NodeNotConnectedException(node, "connection already closed");
    }

    // 2. 根据请求类型选择通道（RECOVERY, BULK, REG, STATE, PING）
    TcpChannel channel = channel(options.type());

    // 3. 确定是否需要压缩
    final Compression.Scheme schemeToUse = getCompressionScheme(request);

    // 4. 通过出站处理器发送
    outboundHandler.sendRequest(
        node,
        channel,
        requestId,
        action,
        request,
        options,
        getTransportVersion(),
        schemeToUse,
        false  // isHandshake = false
    );
}

private Compression.Scheme getCompressionScheme(TransportRequest request) {
    // 全局压缩 或 索引数据压缩
    final boolean shouldCompress = compress == Compression.Enabled.TRUE
        || (compress == Compression.Enabled.INDEXING_DATA
            && request instanceof RawIndexingDataTransportRequest
            && ((RawIndexingDataTransportRequest) request).isRawIndexingData());
    return shouldCompress ? compressionScheme : null;
}

// server/src/main/java/org/elasticsearch/transport/OutboundHandler.java
void sendRequest(
    final DiscoveryNode node,
    final TcpChannel channel,
    final long requestId,
    final String action,
    final TransportRequest request,
    final TransportRequestOptions options,
    final TransportVersion transportVersion,
    final Compression.Scheme compressionScheme,
    final boolean isHandshake
) throws IOException, TransportException {
    // 调用通用发送逻辑
    sendMessage(
        channel,
        MessageDirection.REQUEST,
        action,
        request,
        requestId,
        isHandshake,
        compressionScheme,
        transportVersion,
        ResponseStatsConsumer.NONE,
        () -> messageListener.onRequestSent(node, requestId, action, request, options)
    );
}

private void sendMessage(
    TcpChannel channel,
    MessageDirection messageDirection,
    String action,
    Writeable writeable,
    long requestId,
    boolean isHandshake,
    Compression.Scheme compressionScheme,
    TransportVersion version,
    ResponseStatsConsumer responseStatsConsumer,
    Releasable onAfter
) throws IOException {
    // 1. 序列化消息（包含头部和载荷）
    final BytesReference message;
    final RecyclerBytesStreamOutput byteStreamOutput = new RecyclerBytesStreamOutput(recycler);
    try {
        message = serialize(
            messageDirection,
            action,
            requestId,
            isHandshake,
            version,
            compressionScheme,
            writeable,
            threadPool.getThreadContext(),
            byteStreamOutput
        );
    } catch (Exception e) {
        logger.warn(() -> "failed to serialize outbound message [" + writeable + "]", e);
        throw e;
    }

    // 2. 通过内部发送方法写入通道
    internalSend(channel, message, () -> writeable.getClass().getName(), listener);
}

public static BytesReference serialize(
    MessageDirection messageDirection,
    String action,
    long requestId,
    boolean isHandshake,
    TransportVersion version,
    Compression.Scheme compressionScheme,
    Writeable writeable,
    ThreadContext threadContext,
    RecyclerBytesStreamOutput byteStreamOutput
) throws IOException {
    // 1. 跳过头部空间（稍后填充）
    byteStreamOutput.skip(TcpHeader.HEADER_SIZE);

    // 2. 写入线程上下文（tracing、headers 等）
    threadContext.writeTo(byteStreamOutput);

    // 3. 对于请求，写入 action 名称
    if (messageDirection == MessageDirection.REQUEST) {
        byteStreamOutput.writeString(action);
    }

    // 4. 序列化消息体（可能压缩）
    final int variableHeaderLength = Math.toIntExact(byteStreamOutput.position() - TcpHeader.HEADER_SIZE);
    BytesReference message = serializeMessageBody(writeable, compressionScheme, version, byteStreamOutput);

    // 5. 构建状态标志
    byte status = 0;
    if (messageDirection != MessageDirection.REQUEST) {
        status = TransportStatus.setResponse(status);
    }
    if (isHandshake) {
        status = TransportStatus.setHandshake(status);
    }
    if (compressionScheme != null) {
        status = TransportStatus.setCompress(status);
    }

    // 6. 回填头部
    byteStreamOutput.seek(0);
    TcpHeader.writeHeader(
        byteStreamOutput,
        requestId,
        status,
        version,
        message.length() - TcpHeader.HEADER_SIZE,
        variableHeaderLength
    );

    return message;
}

private void internalSend(
    TcpChannel channel,
    BytesReference reference,
    Supplier<String> messageDescription,
    ActionListener<Void> listener
) {
    // 1. 记录访问时间
    final long startTime = threadPool.rawRelativeTimeInMillis();
    channel.getChannelStats().markAccessed(startTime);

    // 2. 记录发送字节数
    final long messageSize = reference.length();

    // 3. 日志记录（如果启用）
    TransportLogger.logOutboundMessage(channel, reference);

    // 4. 清空线程上下文，写入 Netty 通道
    try (var ignored = threadPool.getThreadContext().newEmptyContext()) {
        channel.sendMessage(reference, new ActionListener<>() {
            @Override
            public void onResponse(Void v) {
                statsTracker.markBytesWritten(messageSize);
                listener.onResponse(v);
            }

            @Override
            public void onFailure(Exception e) {
                logger.warn(() -> "send message failed [channel: " + channel + "]", e);
                listener.onFailure(e);
            }
        });
    }
}

// server/src/main/java/org/elasticsearch/transport/InboundHandler.java
void inboundMessage(TcpChannel channel, InboundMessage message) throws Exception {
    // 1. 记录访问时间
    final long startTime = threadPool.rawRelativeTimeInMillis();
    channel.getChannelStats().markAccessed(startTime);

    // 2. 日志记录
    TransportLogger.logInboundMessage(channel, message);

    // 3. 处理消息（自动关闭）
    if (message.isPing()) {
        keepAlive.receiveKeepAlive(channel);
    } else {
        messageReceived(channel, message, startTime);
    }
}

private void messageReceived(TcpChannel channel, InboundMessage message, long startTime) throws IOException {
    final InetSocketAddress remoteAddress = channel.getRemoteAddress();
    final Header header = message.getHeader();

    ThreadContext threadContext = threadPool.getThreadContext();
    try (var ignored = threadContext.newStoredContext()) {
        // 1. 设置线程上下文（从消息头部恢复）
        threadContext.setHeaders(header.getHeaders());
        threadContext.putTransient("_remote_address", remoteAddress);

        // 2. 根据消息类型分发
        if (header.isRequest()) {
            handleRequest(channel, message);
        } else {
            // 处理响应
            TransportResponseHandler<?> responseHandler = responseHandlers.onResponseReceived(header.getRequestId());
            if (responseHandler != null) {
                executeResponseHandler(message, responseHandler, remoteAddress);
            }
        }
    } finally {
        // 3. 记录慢请求
        final long took = threadPool.rawRelativeTimeInMillis() - startTime;
        if (slowLogThresholdMs > 0 && took > slowLogThresholdMs) {
            logSlowMessage(message, took, slowLogThresholdMs, responseHandler);
        }
    }
}

private <T extends TransportRequest> void handleRequest(TcpChannel channel, InboundMessage message) throws IOException {
    final Header header = message.getHeader();

    // 1. 处理握手请求
    if (header.isHandshake()) {
        handleHandshakeRequest(channel, message);
        return;
    }

    // 2. 获取 action 和 requestId
    final String action = header.getActionName();
    final long requestId = header.getRequestId();

    // 3. 查找请求处理器
    final RequestHandlerRegistry<T> reg = requestHandlers.getHandler(action);

    // 4. 创建传输通道（用于发送响应）
    final TransportChannel transportChannel = new TcpTransportChannel(
        outboundHandler,
        channel,
        action,
        requestId,
        header.getVersion(),
        header.getCompressionScheme(),
        reg,
        false,
        message.takeBreakerReleaseControl()
    );

    try {
        // 5. 触发消息监听器
        messageListener.onRequestReceived(requestId, action);

        // 6. 反序列化请求
        final StreamInput stream = namedWriteableStream(message.openOrGetStreamInput());
        T request = reg.newRequest(stream);
        request.remoteAddress(channel.getRemoteAddress());
        request.setRequestId(requestId);

        // 7. 根据线程池类型执行
        if (reg.getExecutor() == EsExecutors.DIRECT_EXECUTOR_SERVICE) {
            // 直接在当前线程执行
            doHandleRequest(reg, request, transportChannel);
        } else {
            // 提交到线程池执行
            handleRequestForking(request, reg, transportChannel);
        }
    } catch (Exception e) {
        sendErrorResponse(action, transportChannel, e);
    }
}

private static <T extends TransportRequest> void doHandleRequest(
    RequestHandlerRegistry<T> reg,
    T request,
    TransportChannel channel
) {
    try {
        reg.processMessageReceived(request, channel);
    } catch (Exception e) {
        sendErrorResponse(reg.getAction(), channel, e);
    }
}

// server/src/main/java/org/elasticsearch/search/SearchService.java
// （注册的 Handler 在 SearchTransportService 中）
transportService.registerRequestHandler(
    QUERY_ACTION_NAME,  // "indices:data/read/search[phase/query]"
    EsExecutors.DIRECT_EXECUTOR_SERVICE,
    ShardSearchRequest::new,
    (request, channel, task) -> searchService.executeQueryPhase(
        request,
        (SearchShardTask) task,
        new ChannelActionListener<>(channel)
    )
);

public void executeQueryPhase(
    ShardSearchRequest request,
    SearchShardTask task,
    ActionListener<SearchPhaseResult> listener
) {
    // 1. 创建搜索上下文
    final SearchContext searchContext = createContext(request, task);

    try {
        // 2. 执行查询阶段
        searchContext.preProcess();
        QueryPhase.execute(searchContext);

        // 3. 构建查询结果
        QuerySearchResult queryResult = searchContext.queryResult();

        // 4. 返回结果
        listener.onResponse(queryResult);
    } catch (Exception e) {
        listener.onFailure(e);
    } finally {
        searchContext.close();
    }
}

// server/src/main/java/org/elasticsearch/transport/TcpTransportChannel.java
@Override
public void sendResponse(TransportResponse response) {
    try {
        outboundHandler.sendResponse(
            version,
            channel,
            requestId,
            action,
            response,
            compressionScheme,
            false  // isHandshake
        );
    } catch (Exception e) {
        throw new SendRequestTransportException(channel.getRemoteAddress(), action, e);
    }
}

sequenceDiagram
    autonumber
    participant SearchAction as TransportSearchAction<br/>(协调节点)
    participant SearchTransport as SearchTransportService
    participant TS as TransportService
    participant CM as ConnectionManager
    participant Conn as TcpTransport.Connection
    participant Out as OutboundHandler
    participant Channel as TcpChannel/Netty

    Note over SearchAction: Query 阶段开始

    SearchAction->>SearchTransport: sendExecuteQuery(connection, request, task, listener)
    SearchTransport->>SearchTransport: 确定响应类型<br/>(单分片优化)
    SearchTransport->>TS: sendChildRequest(connection, QUERY_ACTION_NAME, request, task, handler)

    TS->>TS: 设置 parentTask
    TS->>TS: sendRequest(connection, action, request, options, handler)
    TS->>TS: asyncSender.sendRequest()<br/>(拦截器处理)

    TS->>TS: sendRequestInternal()
    TS->>TS: 生成 requestId
    TS->>TS: responseHandlers.add(requestId, handler)
    TS->>Conn: connection.sendRequest(requestId, action, request, options)

    Conn->>Conn: 选择通道 channel(type)
    Conn->>Conn: 判断是否压缩
    Conn->>Out: outboundHandler.sendRequest(node, channel, requestId, action, request, ...)

    Out->>Out: sendMessage(channel, REQUEST, action, request, requestId, ...)
    Out->>Out: serialize(REQUEST, action, requestId, request, ...)
    Out->>Out: 1. 跳过头部空间
    Out->>Out: 2. 写入 ThreadContext
    Out->>Out: 3. 写入 action 名称
    Out->>Out: 4. 序列化 request 体
    Out->>Out: 5. 可选压缩
    Out->>Out: 6. 回填头部

    Out->>Channel: channel.sendMessage(bytesRef, listener)
    Channel->>Channel: Netty write & flush

    Note over Channel: 网络传输

    Note over SearchAction,Channel: 数据节点接收和处理

sequenceDiagram
    autonumber
    participant Channel as TcpChannel/Netty<br/>(数据节点)
    participant In as InboundHandler
    participant RH as RequestHandlerRegistry
    participant SearchService as SearchService
    participant QueryPhase as QueryPhase
    participant TChannel as TcpTransportChannel
    participant Out as OutboundHandler

    Note over Channel: 数据节点接收请求

    Channel->>In: inboundMessage(channel, message)
    In->>In: 记录访问时间
    In->>In: 日志记录
    In->>In: messageReceived(channel, message, startTime)

    In->>In: 恢复 ThreadContext
    In->>In: handleRequest(channel, message)
    In->>In: 解析 header (action, requestId)
    In->>RH: requestHandlers.getHandler(action)
    RH-->>In: RequestHandlerRegistry

    In->>In: 创建 TcpTransportChannel
    In->>In: 反序列化 request
    In->>In: 判断执行器类型

    alt 直接执行 (DIRECT_EXECUTOR_SERVICE)
        In->>RH: doHandleRequest(reg, request, channel)
        RH->>SearchService: processMessageReceived(request, channel)<br/>executeQueryPhase(request, task, listener)
    else 线程池执行
        In->>RH: handleRequestForking(request, reg, channel)
        RH->>SearchService: [异步] processMessageReceived(request, channel)
    end

    SearchService->>SearchService: 创建 SearchContext
    SearchService->>QueryPhase: QueryPhase.execute(searchContext)
    QueryPhase->>QueryPhase: 1. runQuery(searchContext)
    QueryPhase->>QueryPhase: 2. executeInternal(searchContext)
    QueryPhase->>QueryPhase: 3. 收集 TopDocs
    QueryPhase-->>SearchService: QuerySearchResult

    SearchService->>TChannel: listener.onResponse(queryResult)
    TChannel->>Out: channel.sendResponse(queryResult)
    Out->>Out: sendResponse(version, channel, requestId, action, response, ...)
    Out->>Out: sendMessage(channel, RESPONSE, action, response, requestId, ...)
    Out->>Out: serialize(RESPONSE, ..., response, ...)
    Out->>Channel: channel.sendMessage(bytesRef, listener)
    Channel->>Channel: Netty write & flush

    Note over Channel: 响应返回协调节点

sequenceDiagram
    autonumber
    participant Channel as TcpChannel/Netty<br/>(协调节点)
    participant In as InboundHandler
    participant ResponseHandler as TransportResponseHandler<br/>(ActionListener)
    participant SearchAction as TransportSearchAction

    Note over Channel: 协调节点接收响应

    Channel->>In: inboundMessage(channel, responseMessage)
    In->>In: messageReceived(channel, responseMessage, startTime)
    In->>In: handleResponse(channel, message)
    In->>In: 解析 header (requestId, status)
    In->>In: responseHandlers.onResponseReceived(requestId)
    In-->>ResponseHandler: 获取 ResponseHandler

    In->>In: executeResponseHandler(message, responseHandler, remoteAddress)
    In->>In: 反序列化 response
    In->>ResponseHandler: handler.handleResponse(response)

    ResponseHandler->>SearchAction: onResponse(queryResult)
    SearchAction->>SearchAction: 聚合结果<br/>继续 Fetch 阶段或返回

    Note over SearchAction: Query 阶段完成

flowchart TD
    A[TransportBulkAction] --> B[TransportShardBulkAction]
    B --> C[TransportReplicationAction]
    C --> D[ReroutePhase: 路由到主分片]
    D --> E[TransportService.sendRequest 主分片]
    E --> F[PrimaryPhase: 主分片执行]
    F --> G[ReplicationPhase: 发送到副本]
    G --> H[ReplicasProxy.performOn]
    H --> I[TransportService.sendRequest 副本]
    I --> J[ReplicaPhase: 副本执行]
    J --> K[等待 Quorum 确认]
    K --> L[返回响应]

// server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java
public TransportReplicationAction(
    Settings settings,
    String actionName,
    TransportService transportService,
    ClusterService clusterService,
    IndicesService indicesService,
    ThreadPool threadPool,
    ShardStateAction shardStateAction,
    ActionFilters actionFilters,
    Writeable.Reader<Request> requestReader,
    Writeable.Reader<ReplicaRequest> replicaRequestReader,
    Executor executor,
    SyncGlobalCheckpointAfterOperation syncGlobalCheckpointAfterOperation,
    PrimaryActionExecution primaryActionExecution,
    ReplicaActionExecution replicaActionExecution
) {
    // ...
    this.transportPrimaryAction = actionName + "[p]";  // 主分片 action
    this.transportReplicaAction = actionName + "[r]";   // 副本分片 action

    // 1. 注册主请求处理器（入口）
    transportService.registerRequestHandler(
        actionName,
        EsExecutors.DIRECT_EXECUTOR_SERVICE,
        requestReader,
        this::handleOperationRequest
    );

    // 2. 注册主分片处理器
    transportService.registerRequestHandler(
        transportPrimaryAction,
        executor,
        forceExecutionOnPrimary,
        true,  // canTripCircuitBreaker
        in -> new ConcreteShardRequest<>(requestReader, in),
        this::handlePrimaryRequest
    );

    // 3. 注册副本分片处理器
    transportService.registerRequestHandler(
        transportReplicaAction,
        executor,
        true,  // forceExecution - 副本不能拒绝
        canTripCircuitBreakerOnReplica,
        in -> new ConcreteReplicaRequest<>(replicaRequestReader, in),
        this::handleReplicaRequest
    );
}