# 创建索引
PUT /my-index
{
  "mappings": {
    "properties": {
      "title": { "type": "text" },
      "content": { "type": "text" },
      "timestamp": { "type": "date" }
    }
  }
}

# 索引文档
POST /my-index/_doc/1
{
  "title": "Elasticsearch Guide",
  "content": "This is a comprehensive guide to Elasticsearch",
  "timestamp": "2024-01-01T00:00:00Z"
}

# 搜索文档
GET /my-index/_search
{
  "query": {
    "match": {
      "title": "Elasticsearch"
    }
  }
}

graph TB
    subgraph "Client Layer"
        A[REST Client]
        B[Java Client]
        C[Other Language Clients]
    end

    subgraph "REST Layer"
        D[RestController]
        E[RestHandler]
        F[ActionModule]
    end

    subgraph "Transport Layer"
        G[TransportService]
        H[ActionListener]
        I[ThreadPool]
    end

    subgraph "Core Services"
        J[SearchService]
        K[IndexService]
        L[ClusterService]
        M[IndicesService]
    end

    subgraph "Storage Layer"
        N[Lucene Engine]
        O[TransLog]
        P[IndexShard]
        Q[FileSystem]
    end

    A --> D
    B --> G
    C --> D
    D --> E
    E --> F
    F --> G
    G --> H
    H --> I
    I --> J
    I --> K
    I --> L
    J --> M
    K --> M
    L --> M
    M --> N
    M --> O
    M --> P
    N --> Q
    O --> Q
    P --> Q

sequenceDiagram
    participant Client
    participant RestController
    participant TransportAction
    participant ClusterService
    participant IndexService
    participant Lucene

    Client->>RestController: HTTP Request
    RestController->>RestController: Route Resolution
    RestController->>TransportAction: Execute Action
    TransportAction->>ClusterService: Get Cluster State
    ClusterService-->>TransportAction: Cluster Metadata
    TransportAction->>IndexService: Process Request
    IndexService->>Lucene: Execute Operation
    Lucene-->>IndexService: Operation Result
    IndexService-->>TransportAction: Response
    TransportAction-->>RestController: Action Response
    RestController-->>Client: HTTP Response

graph TD
    A[HTTP Request] --> B[RestController]
    B --> C[Route Resolution]
    C --> D[RestHandler Selection]
    D --> E[Parameter Parsing]
    E --> F[Request Validation]
    F --> G[Transport Action]
    G --> H[Response Generation]
    H --> I[HTTP Response]

public class RestController implements HttpServerTransport.Dispatcher {

    // 路径树，用于快速路由匹配
    private final PathTrie<MethodHandlers> handlers = new PathTrie<>(RestUtils.REST_DECODER);

    // 请求拦截器
    private final RestInterceptor interceptor;

    // 节点客户端
    private final NodeClient client;

    /**
     * 核心请求分发方法
     * 功能：接收HTTP请求，解析路径，找到对应的处理器并执行
     */
    @Override
    public void dispatchRequest(RestRequest request, RestChannel channel, ThreadContext threadContext) {
        // 添加Elasticsearch产品标识头
        threadContext.addResponseHeader(ELASTIC_PRODUCT_HTTP_HEADER, ELASTIC_PRODUCT_HTTP_HEADER_VALUE);
        try {
            // 尝试所有可能的处理器
            tryAllHandlers(request, channel, threadContext);
        } catch (Exception e) {
            try {
                sendFailure(channel, e);
            } catch (Exception inner) {
                inner.addSuppressed(e);
                logger.error(() -> "failed to send failure response for uri [" + request.uri() + "]", inner);
            }
        }
    }

    /**
     * 注册REST处理器
     * 功能：将REST端点与处理器关联
     */
    public void registerHandler(final RestHandler handler) {
        handler.routes().forEach(route -> registerHandler(route, handler));
    }
}

public class ActionModule extends AbstractModule {

    /**
     * 初始化REST处理器
     * 功能：注册所有内置的REST API端点
     */
    private void initRestHandlers(Consumer<RestHandler> registerHandler) {
        // 索引操作相关API
        registerHandler.accept(new RestIndexAction(clusterService, projectIdResolver));
        registerHandler.accept(new RestGetAction());
        registerHandler.accept(new RestDeleteAction());
        registerHandler.accept(new RestUpdateAction());

        // 搜索操作相关API
        registerHandler.accept(new RestSearchAction(restController.getSearchUsageHolder(),
                                                   clusterSupportsFeature, settings));
        registerHandler.accept(new RestMultiSearchAction(settings,
                                                        restController.getSearchUsageHolder(),
                                                        clusterSupportsFeature));

        // 批量操作API
        registerHandler.accept(new RestBulkAction(settings, clusterSettings, bulkService));

        // 集群管理API
        registerHandler.accept(new RestClusterHealthAction());
        registerHandler.accept(new RestClusterStateAction());

        // 索引管理API
        registerHandler.accept(new RestCreateIndexAction());
        registerHandler.accept(new RestDeleteIndexAction());
        registerHandler.accept(new RestGetMappingAction());
        registerHandler.accept(new RestPutMappingAction());
    }
}

RestSearchAction.prepareRequest()
→ parseSearchRequest()
→ TransportSearchAction.execute()
→ SearchService.executeQueryPhase()
→ Lucene查询执行

RestIndexAction.prepareRequest()
→ IndexRequest构建
→ TransportIndexAction.execute()
→ IndexService.index()
→ Lucene文档索引

RestBulkAction.prepareRequest()
→ BulkRequest解析
→ TransportBulkAction.execute()
→ BulkOperation.executeBulkShardRequests()
→ 分片级批量处理

graph TB
    subgraph "Search Request Processing"
        A[SearchRequest] --> B[SearchSourceBuilder]
        B --> C[Query Parsing]
        C --> D[Search Context]
    end

    subgraph "Search Execution"
        E[SearchService] --> F[SearchPhase]
        F --> G[QueryPhase]
        F --> H[FetchPhase]
        G --> I[DfsPhase]
    end

    subgraph "Shard Level Processing"
        J[ShardSearchRequest] --> K[SearchContext]
        K --> L[Lucene Query]
        L --> M[TopDocs]
        M --> N[SearchHits]
    end

    D --> E
    E --> J
    N --> H

public class SearchRequest extends LegacyActionRequest implements IndicesRequest.Replaceable, Rewriteable<SearchRequest> {

    // 搜索类型：QUERY_THEN_FETCH 或 DFS_QUERY_THEN_FETCH
    private SearchType searchType = SearchType.DEFAULT;

    // 目标索引
    private String[] indices = Strings.EMPTY_ARRAY;

    // 路由参数
    private String routing;

    // 搜索偏好设置
    private String preference;

    // 搜索源构建器
    private SearchSourceBuilder source;

    // 请求缓存设置
    private Boolean requestCache;

    // 是否允许部分搜索结果
    private Boolean allowPartialSearchResults;

    // 滚动搜索保持时间
    private TimeValue scrollKeepAlive;

    // 批量reduce大小
    private int batchedReduceSize = DEFAULT_BATCHED_REDUCE_SIZE;

    // 最大并发分片请求数
    private int maxConcurrentShardRequests = 0;

    /**
     * 设置搜索源
     */
    public SearchRequest source(SearchSourceBuilder sourceBuilder) {
        this.source = sourceBuilder;
        return this;
    }

    /**
     * 设置目标索引
     */
    public SearchRequest indices(String... indices) {
        this.indices = indices;
        return this;
    }
}

public class TransportSearchAction extends HandledTransportAction<SearchRequest, SearchResponse> {

    /**
     * 执行搜索请求的核心方法
     */
    private void executeRequest(
        SearchTask task,
        SearchRequest original,
        ActionListener<SearchResponse> originalListener,
        Function<ActionListener<SearchResponse>, SearchPhaseProvider> searchPhaseProvider,
        boolean collectSearchTelemetry
    ) {
        // 创建时间提供器
        final long relativeStartNanos = System.nanoTime();
        final SearchTimeProvider timeProvider = new SearchTimeProvider(
            original.getOrCreateAbsoluteStartMillis(),
            relativeStartNanos,
            System::nanoTime
        );

        // 获取集群状态
        final ClusterState clusterState = clusterService.state();
        clusterState.blocks().globalBlockedRaiseException(projectResolver.getProjectId(), ClusterBlockLevel.READ);

        // 解析目标索引
        ProjectState projectState = projectResolver.getProjectState(clusterState);
        final ResolvedIndices resolvedIndices;
        if (original.pointInTimeBuilder() != null) {
            // 使用Point in Time解析
            resolvedIndices = ResolvedIndices.resolveWithPIT(
                original.pointInTimeBuilder(),
                original.indicesOptions(),
                projectState.metadata(),
                namedWriteableRegistry
            );
        } else {
            // 常规索引解析
            resolvedIndices = ResolvedIndices.resolveWithIndicesRequest(
                original,
                projectState.metadata(),
                indexNameExpressionResolver,
                remoteClusterService,
                timeProvider.absoluteStartMillis()
            );
            frozenIndexCheck(resolvedIndices);
        }

        // 执行搜索阶段
        executeSearch(task, timeProvider, original, originalListener,
                     clusterState, resolvedIndices, searchPhaseProvider,
                     collectSearchTelemetry);
    }
}

public class SearchService extends AbstractLifecycleComponent implements IndexEventListener {

    /**
     * 执行查询阶段
     */
    public void executeQueryPhase(ShardSearchRequest request, SearchShardTask task, ActionListener<SearchPhaseResult> listener) {
        final IndexService indexService = indicesService.indexServiceSafe(request.shardId().getIndex());
        final IndexShard indexShard = indexService.getShard(request.shardId().getId());

        // 创建搜索上下文
        final SearchContext searchContext = createSearchContext(request, indexShard, task);

        try {
            // 执行查询
            queryPhase.execute(searchContext);

            // 返回查询结果
            if (searchContext.queryResult().hasHits()) {
                listener.onResponse(searchContext.queryResult());
            } else {
                listener.onResponse(new QuerySearchResult());
            }
        } catch (Exception e) {
            listener.onFailure(e);
        } finally {
            // 清理搜索上下文
            cleanupSearchContext(searchContext);
        }
    }

    /**
     * 执行获取阶段
     */
    public void executeFetchPhase(ShardFetchRequest request, SearchShardTask task, ActionListener<FetchSearchResult> listener) {
        final SearchContext searchContext = findSearchContext(request.contextId());

        try {
            // 执行文档获取
            fetchPhase.execute(searchContext);
            listener.onResponse(searchContext.fetchResult());
        } catch (Exception e) {
            listener.onFailure(e);
        }
    }
}

sequenceDiagram
    participant Client
    participant RestSearchAction
    participant TransportSearchAction
    participant SearchService
    participant IndexShard
    participant Lucene

    Client->>RestSearchAction: POST /_search
    RestSearchAction->>RestSearchAction: parseSearchRequest()
    RestSearchAction->>TransportSearchAction: execute(SearchRequest)

    TransportSearchAction->>TransportSearchAction: resolveIndices()
    TransportSearchAction->>SearchService: executeQueryPhase()

    SearchService->>SearchService: createSearchContext()
    SearchService->>IndexShard: acquireSearcher()
    IndexShard->>Lucene: IndexSearcher

    SearchService->>SearchService: queryPhase.execute()
    SearchService->>Lucene: search(Query, TopDocsCollector)
    Lucene-->>SearchService: TopDocs

    alt Fetch Phase Required
        SearchService->>SearchService: executeFetchPhase()
        SearchService->>Lucene: doc(docId)
        Lucene-->>SearchService: Document
    end

    SearchService-->>TransportSearchAction: SearchPhaseResult
    TransportSearchAction-->>RestSearchAction: SearchResponse
    RestSearchAction-->>Client: HTTP Response

graph TB
    subgraph "Index Request Processing"
        A[IndexRequest] --> B[Document Parsing]
        B --> C[Field Mapping]
        C --> D[Document Validation]
    end

    subgraph "Index Execution"
        E[IndexService] --> F[IndexShard]
        F --> G[InternalEngine]
        G --> H[Lucene IndexWriter]
    end

    subgraph "Storage Layer"
        I[TransLog] --> J[Lucene Segments]
        J --> K[File System]
        H --> I
        H --> J
    end

    D --> E
    E --> F
    F --> G

public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implements DocWriteRequest<IndexRequest>, CompositeIndicesRequest {

    // 文档ID
    private String id;

    // 路由参数
    private String routing;

    // 索引源数据
    private final IndexSource indexSource;

    // 操作类型：INDEX 或 CREATE
    private OpType opType = OpType.INDEX;

    // 版本控制
    private long version = Versions.MATCH_ANY;
    private VersionType versionType = VersionType.INTERNAL;

    // 管道处理
    private String pipeline;
    private String finalPipeline;

    /**
     * 设置文档源数据
     */
    public IndexRequest source(Map<String, ?> source, XContentType contentType) throws ElasticsearchGenerationException {
        try {
            XContentBuilder builder = XContentFactory.contentBuilder(contentType);
            builder.map(source);
            return source(builder);
        } catch (IOException e) {
            throw new ElasticsearchGenerationException("Failed to generate [" + source + "]", e);
        }
    }

    /**
     * 设置文档ID
     */
    public IndexRequest id(String id) {
        this.id = id;
        return this;
    }

    /**
     * 设置操作类型
     */
    public IndexRequest opType(OpType opType) {
        this.opType = opType;
        return this;
    }
}

public class TransportIndexAction extends TransportReplicationAction<IndexRequest, IndexRequest, IndexResponse> {

    /**
     * 在主分片上执行索引操作
     */
    @Override
    protected WritePrimaryResult<IndexRequest, IndexResponse> shardOperationOnPrimary(
        IndexRequest request, IndexShard primary, ActionListener<WritePrimaryResult<IndexRequest, IndexResponse>> listener) {

        try {
            // 执行索引操作
            final Engine.IndexResult result = primary.applyIndexOperationOnPrimary(
                request.version(),
                request.versionType(),
                new SourceToParse(
                    request.index(),
                    request.id(),
                    request.source(),
                    request.getContentType(),
                    request.routing()
                ),
                request.getIfSeqNo(),
                request.getIfPrimaryTerm(),
                request.getAutoGeneratedTimestamp(),
                request.isRetry()
            );

            // 创建响应
            final IndexResponse response = new IndexResponse(
                primary.shardId(),
                request.id(),
                result.getSeqNo(),
                result.getTerm(),
                result.getVersion(),
                result.isCreated()
            );

            return new WritePrimaryResult<>(request, response, result.getResultType(), null, primary);

        } catch (Exception e) {
            return new WritePrimaryResult<>(request, e, primary);
        }
    }

    /**
     * 在副本分片上执行索引操作
     */
    @Override
    protected WriteReplicaResult shardOperationOnReplica(IndexRequest request, IndexShard replica) {
        try {
            final Engine.IndexResult result = replica.applyIndexOperationOnReplica(
                request.getSeqNo(),
                request.getPrimaryTerm(),
                request.version(),
                request.getAutoGeneratedTimestamp(),
                request.isRetry(),
                new SourceToParse(
                    request.index(),
                    request.id(),
                    request.source(),
                    request.getContentType(),
                    request.routing()
                )
            );

            return new WriteReplicaResult(request, result.getResultType(), null, replica, logger);

        } catch (Exception e) {
            return new WriteReplicaResult(request, e, replica, logger);
        }
    }
}

public class IndexService extends AbstractIndexComponent implements IndicesClusterStateService.AllocatedIndex<IndexShard> {

    // 索引设置
    private final IndexSettings indexSettings;

    // 分析器注册表
    private final IndexAnalyzers indexAnalyzers;

    // 映射服务
    private final MapperService mapperService;

    // 相似度服务
    private final SimilarityService similarityService;

    // 分片映射
    private volatile Map<Integer, IndexShard> shards = emptyMap();

    /**
     * 创建新的索引分片
     */
    public synchronized IndexShard createShard(
        final ShardRouting routing,
        final PeerRecoveryTargetService recoveryTargetService,
        final PeerRecoveryTargetService.RecoveryListener recoveryListener,
        final RepositoriesService repositoriesService,
        final Consumer<IndexShard.ShardFailure> onShardFailure,
        final Consumer<ShardId> globalCheckpointSyncer,
        final RetentionLeaseSyncer retentionLeaseSyncer,
        final DiscoveryNode targetNode,
        final DiscoveryNode sourceNode,
        final long primaryTerm,
        final IndexEventListener... listeners
    ) throws IOException {

        // 创建分片实例
        final IndexShard indexShard = new IndexShard(
            routing,
            this.indexSettings,
            indexPath,
            store,
            indexSortSupplier,
            indexCache,
            mapperService,
            similarityService,
            engineFactory,
            indexEventListener,
            queryShardContext,
            globalCheckpointSyncer,
            retentionLeaseSyncer,
            primaryTerm,
            Arrays.asList(listeners)
        );

        // 注册分片
        shards = Maps.copyMapWithAddedEntry(shards, shardId, indexShard);

        return indexShard;
    }

    /**
     * 获取映射服务
     */
    public MapperService mapperService() {
        return mapperService;
    }
}

sequenceDiagram
    participant Client
    participant RestIndexAction
    participant TransportIndexAction
    participant IndexService
    participant IndexShard
    participant Engine
    participant Lucene

    Client->>RestIndexAction: PUT /index/_doc/1
    RestIndexAction->>RestIndexAction: parseRequest()
    RestIndexAction->>TransportIndexAction: execute(IndexRequest)

    TransportIndexAction->>TransportIndexAction: resolveIndex()
    TransportIndexAction->>IndexService: getShard()
    IndexService-->>TransportIndexAction: IndexShard

    TransportIndexAction->>IndexShard: applyIndexOperationOnPrimary()
    IndexShard->>Engine: index(SourceToParse)

    Engine->>Engine: parseDocument()
    Engine->>Engine: validateSequenceNumber()
    Engine->>Lucene: IndexWriter.addDocument()

    Lucene-->>Engine: IndexResult
    Engine-->>IndexShard: IndexResult
    IndexShard-->>TransportIndexAction: IndexResult

    alt Replicas Exist
        TransportIndexAction->>IndexShard: applyIndexOperationOnReplica()
        IndexShard->>Engine: index(SourceToParse)
        Engine->>Lucene: IndexWriter.addDocument()
    end

    TransportIndexAction-->>RestIndexAction: IndexResponse
    RestIndexAction-->>Client: HTTP Response

graph TB
    subgraph "Cluster Coordination"
        A[Coordinator] --> B[ElectionStrategy]
        A --> C[JoinHelper]
        A --> D[PublicationTransportHandler]
    end

    subgraph "Cluster State Management"
        E[ClusterService] --> F[MasterService]
        F --> G[ClusterStateTaskExecutor]
        G --> H[ClusterStatePublisher]
    end

    subgraph "Discovery & Membership"
        I[DiscoveryModule] --> J[SeedHostsProvider]
        I --> K[NodeHealthService]
        I --> L[LeaderChecker]
    end

    A --> E
    E --> I

public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {

    // 选举策略
    private final ElectionStrategy electionStrategy;

    // 传输服务
    private final TransportService transportService;

    // 主服务
    private final MasterService masterService;

    // 分配服务
    private final AllocationService allocationService;

    // 加入助手
    private final JoinHelper joinHelper;

    // 协调状态
    private final SetOnce<CoordinationState> coordinationState = new SetOnce<>();

    /**
     * 启动协调器
     */
    @Override
    protected void doStart() {
        synchronized (mutex) {
            // 初始化协调状态
            coordinationState.set(new CoordinationState(
                settings,
                localNode,
                persistedStateSupplier.get(),
                electionStrategy,
                nodeHealthService
            ));

            // 启动预投票收集器
            preVoteCollector.start();

            // 启动领导者检查器
            leaderChecker.start();

            // 启动跟随者检查器
            followersChecker.start();

            // 开始选举过程
            becomeCandidate("coordinator started");
        }
    }

    /**
     * 发布集群状态
     */
    @Override
    public void publish(ClusterStatePublicationEvent clusterStatePublicationEvent,
                       ActionListener<Void> publishListener,
                       AckListener ackListener) {

        final ClusterState clusterState = clusterStatePublicationEvent.getNewState();

        // 验证发布权限
        assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
        assert mode == Mode.LEADER : "not currently leading";

        // 创建发布请求
        final PublishRequest publishRequest = coordinationState.get().handleClientValue(clusterState);

        // 执行发布
        publicationHandler.publish(
            publishRequest,
            wrapWithMutex(publishListener),
            wrapWithMutex(ackListener)
        );
    }

    /**
     * 处理加入请求
     */
    public void handleJoinRequest(JoinRequest joinRequest, ActionListener<Void> listener) {
        synchronized (mutex) {
            if (mode == Mode.LEADER) {
                // 作为领导者处理加入请求
                joinHelper.handleJoinRequest(joinRequest, listener);
            } else {
                // 非领导者拒绝加入请求
                listener.onFailure(new CoordinationStateRejectedException("not the leader"));
            }
        }
    }
}

public class ClusterService extends AbstractLifecycleComponent {

    // 集群设置
    private final ClusterSettings clusterSettings;

    // 主服务
    private final MasterService masterService;

    // 集群应用服务
    private final ClusterApplierService clusterApplierService;

    // 当前集群状态
    private volatile ClusterState clusterState;

    /**
     * 获取当前集群状态
     */
    public ClusterState state() {
        return this.clusterState;
    }

    /**
     * 创建任务队列
     */
    public <T> MasterServiceTaskQueue<T> createTaskQueue(
        String name,
        Priority priority,
        ClusterStateTaskExecutor<T> executor
    ) {
        return masterService.createTaskQueue(name, priority, executor);
    }

    /**
     * 提交集群状态任务
     */
    public <T> void submitStateUpdateTask(
        String source,
        T task,
        ClusterStateTaskConfig config,
        ClusterStateTaskExecutor<T> executor,
        ClusterStateTaskListener listener
    ) {
        masterService.submitStateUpdateTask(source, task, config, executor, listener);
    }

    /**
     * 添加集群状态监听器
     */
    public void addListener(ClusterStateListener listener) {
        clusterApplierService.addListener(listener);
    }

    /**
     * 添加集群状态应用器
     */
    public void addStateApplier(ClusterStateApplier applier) {
        clusterApplierService.addStateApplier(applier);
    }
}

public class ClusterState implements ChunkedToXContent, Diffable<ClusterState> {

    // 集群名称
    private final String clusterName;

    // 状态版本
    private final long version;

    // 状态UUID
    private final String stateUUID;

    // 节点信息
    private final DiscoveryNodes nodes;

    // 元数据
    private final Metadata metadata;

    // 路由表
    private final RoutingTable routingTable;

    // 集群块
    private final ClusterBlocks blocks;

    /**
     * 构建器模式创建集群状态
     */
    public static Builder builder(ClusterState state) {
        return new Builder(state);
    }

    /**
     * 获取主节点
     */
    public DiscoveryNode getMasterNode() {
        return nodes.getMasterNode();
    }

    /**
     * 检查是否有主节点
     */
    public boolean hasMasterNode() {
        return nodes.getMasterNodeId() != null;
    }

    /**
     * 获取索引元数据
     */
    public IndexMetadata getIndexMetadata(String index) {
        return metadata.index(index);
    }

    /**
     * 集群状态构建器
     */
    public static class Builder {
        private String clusterName;
        private long version = 0;
        private String uuid = UNKNOWN_UUID;
        private DiscoveryNodes.Builder nodesBuilder;
        private Metadata.Builder metadataBuilder;
        private RoutingTable.Builder routingTableBuilder;
        private ClusterBlocks.Builder blocksBuilder;

        public Builder nodes(DiscoveryNodes.Builder nodesBuilder) {
            this.nodesBuilder = nodesBuilder;
            return this;
        }

        public Builder metadata(Metadata.Builder metadataBuilder) {
            this.metadataBuilder = metadataBuilder;
            return this;
        }

        public ClusterState build() {
            return new ClusterState(
                clusterName,
                version,
                uuid,
                nodesBuilder.build(),
                metadataBuilder.build(),
                routingTableBuilder.build(),
                blocksBuilder.build()
            );
        }
    }
}

sequenceDiagram
    participant Node1 as Node 1 (Candidate)
    participant Node2 as Node 2
    participant Node3 as Node 3
    participant Cluster as Cluster State

    Note over Node1,Node3: Election Process
    Node1->>Node1: becomeCandidate()
    Node1->>Node2: PreVoteRequest
    Node1->>Node3: PreVoteRequest

    Node2-->>Node1: PreVoteResponse(granted)
    Node3-->>Node1: PreVoteResponse(granted)

    Node1->>Node1: becomeLeader()

    Note over Node1,Node3: State Publication
    Node1->>Cluster: ClusterState Update
    Node1->>Node2: PublishRequest
    Node1->>Node3: PublishRequest

    Node2->>Node2: applyClusterState()
    Node3->>Node3: applyClusterState()

    Node2-->>Node1: PublishResponse(accepted)
    Node3-->>Node1: PublishResponse(accepted)

    Node1->>Node2: CommitRequest
    Node1->>Node3: CommitRequest

    Node2-->>Node1: Acknowledgment
    Node3-->>Node1: Acknowledgment

graph TB
    subgraph "Bulk Request Processing"
        A[BulkRequest] --> B[Request Parsing]
        B --> C[Shard Grouping]
        C --> D[BulkShardRequest]
    end

    subgraph "Bulk Execution"
        E[TransportBulkAction] --> F[BulkOperation]
        F --> G[TransportShardBulkAction]
        G --> H[Engine Operations]
    end

    subgraph "Result Aggregation"
        I[BulkItemResponse] --> J[BulkResponse]
        J --> K[Response Generation]
    end

    D --> E
    H --> I

public class BulkRequest extends LegacyActionRequest
    implements CompositeIndicesRequest, WriteRequest<BulkRequest>, Accountable, RawIndexingDataTransportRequest {

    // 请求列表
    private final List<DocWriteRequest<?>> requests = new ArrayList<>();

    // 涉及的索引集合
    private Set<String> indices = emptySet();

    // 估算的请求大小
    private long sizeInBytes = 0;

    // 全局参数
    private String globalPipeline;
    private String globalRouting;
    private String globalIndex;

    /**
     * 添加索引请求
     */
    public BulkRequest add(IndexRequest request) {
        return internalAdd(request);
    }

    BulkRequest internalAdd(IndexRequest request) {
        Objects.requireNonNull(request, "'request' must not be null");
        applyGlobalMandatoryParameters(request);

        requests.add(request);
        // 计算请求大小
        sizeInBytes += request.indexSource().byteLength() + REQUEST_OVERHEAD;
        indices.add(request.index());
        return this;
    }

    /**
     * 添加更新请求
     */
    public BulkRequest add(UpdateRequest request) {
        return internalAdd(request);
    }

    BulkRequest internalAdd(UpdateRequest request) {
        Objects.requireNonNull(request, "'request' must not be null");
        applyGlobalMandatoryParameters(request);

        requests.add(request);
        if (request.doc() != null) {
            sizeInBytes += request.doc().indexSource().byteLength();
        }
        if (request.upsertRequest() != null) {
            sizeInBytes += request.upsertRequest().indexSource().byteLength();
        }
        if (request.script() != null) {
            sizeInBytes += request.script().getIdOrCode().length() * 2;
        }
        indices.add(request.index());
        return this;
    }

    /**
     * 添加删除请求
     */
    public BulkRequest add(DeleteRequest request) {
        Objects.requireNonNull(request, "'request' must not be null");
        applyGlobalMandatoryParameters(request);

        requests.add(request);
        sizeInBytes += REQUEST_OVERHEAD;
        indices.add(request.index());
        return this;
    }

    /**
     * 获取请求数量
     */
    public int numberOfActions() {
        return requests.size();
    }

    /**
     * 获取估算大小
     */
    public long estimatedSizeInBytes() {
        return sizeInBytes;
    }
}

public class TransportBulkAction extends HandledTransportAction<BulkRequest, BulkResponse> {

    /**
     * 执行批量操作
     */
    @Override
    protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<BulkResponse> listener) {
        final long startTime = relativeTime();
        final AtomicArray<BulkItemResponse> responses = new AtomicArray<>(bulkRequest.numberOfActions());

        // 创建批量操作实例
        BulkOperation bulkOperation = new BulkOperation(
            task,
            bulkRequest,
            listener,
            responses,
            startTime,
            clusterService,
            ingestService,
            executorService,
            relativeTimeProvider
        );

        // 执行批量操作
        bulkOperation.execute();
    }
}

public class BulkOperation {

    private final Task task;
    private final BulkRequest bulkRequest;
    private final ActionListener<BulkResponse> listener;
    private final AtomicArray<BulkItemResponse> responses;

    /**
     * 执行批量操作
     */
    public void execute() {
        final ClusterState clusterState = clusterService.state();

        // 检查集群块
        if (addFailureIfIndexIsUnavailable(clusterState)) {
            return;
        }

        // 解析和验证请求
        final Map<String, List<BulkItemRequest>> requestsByIndex = new HashMap<>();
        for (int i = 0; i < bulkRequest.numberOfActions(); i++) {
            DocWriteRequest<?> docWriteRequest = bulkRequest.requests().get(i);

            // 解析索引名称
            String indexName = docWriteRequest.index();

            // 验证请求
            BulkItemRequest bulkItemRequest = new BulkItemRequest(i, docWriteRequest);
            requestsByIndex.computeIfAbsent(indexName, k -> new ArrayList<>()).add(bulkItemRequest);
        }

        // 按分片分组请求
        final Map<ShardId, List<BulkItemRequest>> requestsByShard = groupRequestsByShard(requestsByIndex, clusterState);

        // 执行分片级批量操作
        executeBulkShardRequests(requestsByShard, clusterState);
    }

    /**
     * 按分片分组请求
     */
    private Map<ShardId, List<BulkItemRequest>> groupRequestsByShard(
        Map<String, List<BulkItemRequest>> requestsByIndex,
        ClusterState clusterState
    ) {
        final Map<ShardId, List<BulkItemRequest>> requestsByShard = new HashMap<>();

        for (Map.Entry<String, List<BulkItemRequest>> entry : requestsByIndex.entrySet()) {
            String indexName = entry.getKey();
            List<BulkItemRequest> requests = entry.getValue();

            IndexMetadata indexMetadata = clusterState.metadata().index(indexName);
            if (indexMetadata == null) {
                continue;
            }

            for (BulkItemRequest request : requests) {
                // 计算分片ID
                ShardId shardId = clusterService.operationRouting().indexShards(
                    clusterState,
                    indexName,
                    request.request().id(),
                    request.request().routing()
                ).shardId();

                requestsByShard.computeIfAbsent(shardId, k -> new ArrayList<>()).add(request);
            }
        }

        return requestsByShard;
    }

    /**
     * 执行分片级批量请求
     */
    private void executeBulkShardRequests(
        Map<ShardId, List<BulkItemRequest>> requestsByShard,
        ClusterState clusterState
    ) {
        ProjectMetadata project = projectResolver.getProjectMetadata(clusterState);
        try (RefCountingRunnable bulkItemRequestCompleteRefCount = new RefCountingRunnable(onRequestsCompleted)) {
            for (Map.Entry<ShardId, List<BulkItemRequest>> entry : requestsByShard.entrySet()) {
                final ShardId shardId = entry.getKey();
                final List<BulkItemRequest> requests = entry.getValue();

                // 创建分片批量请求
                BulkShardRequest bulkShardRequest = new BulkShardRequest(
                    shardId,
                    bulkRequest.getRefreshPolicy(),
                    requests.toArray(new BulkItemRequest[0]),
                    bulkRequest.isSimulated()
                );

                // 设置推理字段映射
                var indexMetadata = project.index(shardId.getIndexName());
                if (indexMetadata != null && indexMetadata.getInferenceFields().isEmpty() == false) {
                    bulkShardRequest.setInferenceFieldMap(indexMetadata.getInferenceFields());
                }

                // 执行分片请求
                executeBulkShardRequest(bulkShardRequest, project.id(), bulkItemRequestCompleteRefCount.acquire());
            }
        }
    }
}