import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
/** end of {@link PingContextProvider } implementation */


    @Override
    public void publish(ClusterChangedEvent clusterChangedEvent, AckListener ackListener) {
        if (!clusterChangedEvent.state().getNodes().isLocalNodeElectedMaster()) {
            throw new IllegalStateException("Shouldn't publish state when not master");
        }
        try {
            publishClusterState.publish(clusterChangedEvent, electMaster.minimumMasterNodes(), ackListener);
        } catch (FailedToCommitClusterStateException t) {
            // cluster service logs a WARN message
            logger.debug("failed to publish cluster state version [{}] (not enough nodes acknowledged, min master nodes [{}])", clusterChangedEvent.state().version(), electMaster.minimumMasterNodes());
            submitRejoin("zen-disco-failed-to-publish");
            throw t;
        }

        // update the set of nodes to ping after the new cluster state has been published
        nodesFD.updateNodesAndPing(clusterChangedEvent.state());

        // clean the pending cluster queue - we are currently master, so any pending cluster state should be failed
        // note that we also clean the queue on master failure (see handleMasterGone) but a delayed cluster state publish
        // from a stale master can still make it in the queue during the election (but not be committed)
        publishClusterState.pendingStatesQueue().failAllStatesAndClear(new ElasticsearchException("elected as master"));
    }
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
@Override
public void applyClusterState(ClusterChangedEvent event) {
    try {
        if (event.localNodeMaster()) {
            if (event.nodesRemoved()) {
                processSnapshotsOnRemovedNodes(event);
            }
            if (event.routingTableChanged()) {
                processStartedShards(event);
            }
            finalizeSnapshotDeletionFromPreviousMaster(event);
        }
    } catch (Exception e) {
        logger.warn("Failed to update snapshot state ", e);
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
private boolean waitingShardsStartedOrUnassigned(ClusterChangedEvent event) {
    SnapshotsInProgress curr = event.state().custom(SnapshotsInProgress.TYPE);
    if (curr != null) {
        for (SnapshotsInProgress.Entry entry : curr.entries()) {
            if (entry.state() == State.STARTED && !entry.waitingIndices().isEmpty()) {
                for (ObjectCursor<String> index : entry.waitingIndices().keys()) {
                    if (event.indexRoutingTableChanged(index.value)) {
                        IndexRoutingTable indexShardRoutingTable = event.state().getRoutingTable().index(index.value);
                        for (ShardId shardId : entry.waitingIndices().get(index.value)) {
                            ShardRouting shardRouting = indexShardRoutingTable.shard(shardId.id()).primaryShard();
                            if (shardRouting != null && (shardRouting.started() || shardRouting.unassigned())) {
                                return true;
                            }
                        }
                    }
                }
            }
        }
    }
    return false;
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
private boolean removedNodesCleanupNeeded(ClusterChangedEvent event) {
    SnapshotsInProgress snapshotsInProgress = event.state().custom(SnapshotsInProgress.TYPE);
    if (snapshotsInProgress == null) {
        return false;
    }
    // Check if we just became the master
    boolean newMaster = !event.previousState().nodes().isLocalNodeElectedMaster();
    for (SnapshotsInProgress.Entry snapshot : snapshotsInProgress.entries()) {
        if (newMaster && (snapshot.state() == State.SUCCESS || snapshot.state() == State.INIT)) {
            // We just replaced old master and snapshots in intermediate states needs to be cleaned
            return true;
        }
        for (DiscoveryNode node : event.nodesDelta().removedNodes()) {
            for (ObjectCursor<ShardSnapshotStatus> shardStatus : snapshot.shards().values()) {
                if (!shardStatus.value.state().completed() && node.getId().equals(shardStatus.value.nodeId())) {
                    // At least one shard was running on the removed node - we need to fail it
                    return true;
                }
            }
        }
    }
    return false;
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
private void cleanupRestoreState(ClusterChangedEvent event) {
    ClusterState state = event.state();

    RestoreInProgress restoreInProgress = state.custom(RestoreInProgress.TYPE);
    if (restoreInProgress != null) {
        for (RestoreInProgress.Entry entry : restoreInProgress.entries()) {
            if (entry.state().completed()) {
                assert completed(entry.shards()) : "state says completed but restore entries are not";
                clusterService.submitStateUpdateTask(
                    "clean up snapshot restore state",
                    new CleanRestoreStateTaskExecutor.Task(entry.snapshot()),
                    ClusterStateTaskConfig.build(Priority.URGENT),
                    cleanRestoreStateTaskExecutor,
                    cleanRestoreStateTaskExecutor);
            }
        }
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
@Override
public void applyClusterState(ClusterChangedEvent event) {
    try {
        SnapshotsInProgress prev = event.previousState().custom(SnapshotsInProgress.TYPE);
        SnapshotsInProgress curr = event.state().custom(SnapshotsInProgress.TYPE);

        if ((prev == null && curr != null) || (prev != null && prev.equals(curr) == false)) {
            processIndexShardSnapshots(event);
        }
        String masterNodeId = event.state().nodes().getMasterNodeId();
        if (masterNodeId != null && masterNodeId.equals(event.previousState().nodes().getMasterNodeId()) == false) {
            syncShardStatsOnNewMaster(event);
        }

    } catch (Exception e) {
        logger.warn("Failed to update snapshot state ", e);
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
public void assertState(ClusterChangedEvent event,
                        boolean stateInMemory,
                        boolean expectMetaData) throws Exception {
    MetaData inMemoryMetaData = null;
    Set<Index> oldIndicesList = emptySet();
    if (stateInMemory) {
        inMemoryMetaData = event.previousState().metaData();
        oldIndicesList = GatewayMetaState.getRelevantIndices(event.previousState(), event.previousState(), oldIndicesList);
    }
    Set<Index> newIndicesList = GatewayMetaState.getRelevantIndices(event.state(),event.previousState(), oldIndicesList);
    // third, get the actual write info
    Iterator<GatewayMetaState.IndexMetaWriteInfo> indices = GatewayMetaState.resolveStatesToBeWritten(oldIndicesList, newIndicesList,
            inMemoryMetaData, event.state().metaData()).iterator();

    if (expectMetaData) {
        assertThat(indices.hasNext(), equalTo(true));
        assertThat(indices.next().getNewMetaData().getIndex().getName(), equalTo("test"));
        assertThat(indices.hasNext(), equalTo(false));
    } else {
        assertThat(indices.hasNext(), equalTo(false));
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
@Override
public void clusterChanged(ClusterChangedEvent event) {
    if (!event.metaDataChanged()) {
        return;
    }

    Set<String> newCurrentSchemas = getNewCurrentSchemas(event.state().metaData());
    synchronized (schemas) {
        Sets.SetView<String> nonBuiltInSchemas = Sets.difference(schemas.keySet(), builtInSchemas.keySet());
        Set<String> deleted = Sets.difference(nonBuiltInSchemas, newCurrentSchemas).immutableCopy();
        Set<String> added = Sets.difference(newCurrentSchemas, schemas.keySet()).immutableCopy();

        for (String deletedSchema : deleted) {
            try {
                schemas.remove(deletedSchema).close();
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
        }
        for (String addedSchema : added) {
            schemas.put(addedSchema, getCustomSchemaInfo(addedSchema));
        }
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
private void applyNewIndices(final ClusterChangedEvent event) {
    // we only create indices for shards that are allocated
    RoutingNodes.RoutingNodeIterator routingNode = event.state().getRoutingNodes().routingNodeIter(event.state().nodes().localNodeId());
    if (routingNode == null) {
        return;
    }
    for (ShardRouting shard : routingNode) {
        if (!indicesService.hasIndex(shard.index())) {
            final IndexMetaData indexMetaData = event.state().metaData().index(shard.index());
            if (logger.isDebugEnabled()) {
                logger.debug("[{}] creating index", indexMetaData.getIndex());
            }
            try {
                indicesService.createIndex(indexMetaData.getIndex(), indexMetaData.getSettings(), event.state().nodes().localNode().id());
            } catch (Throwable e) {
                sendFailShard(shard, indexMetaData.getIndexUUID(), "failed to create index", e);
            }
        }
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
private void applySettings(ClusterChangedEvent event) {
    if (!event.metaDataChanged()) {
        return;
    }
    for (IndexMetaData indexMetaData : event.state().metaData()) {
        if (!indicesService.hasIndex(indexMetaData.getIndex())) {
            // we only create / update here
            continue;
        }
        // if the index meta data didn't change, no need check for refreshed settings
        if (!event.indexMetaDataChanged(indexMetaData)) {
            continue;
        }
        String index = indexMetaData.getIndex();
        IndexService indexService = indicesService.indexService(index);
        if (indexService == null) {
            // already deleted on us, ignore it
            continue;
        }
        IndexSettingsService indexSettingsService = indexService.injector().getInstance(IndexSettingsService.class);
        indexSettingsService.refreshSettings(indexMetaData.getSettings());
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
private void applyAliases(ClusterChangedEvent event) {
    // check if aliases changed
    if (aliasesChanged(event)) {
        // go over and update aliases
        for (IndexMetaData indexMetaData : event.state().metaData()) {
            String index = indexMetaData.getIndex();
            IndexService indexService = indicesService.indexService(index);
            if (indexService == null) {
                // we only create / update here
                continue;
            }
            IndexAliasesService indexAliasesService = indexService.aliasesService();
            indexAliasesService.setAliases(indexMetaData.getAliases());
        }
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
@Override
public void publish(ClusterChangedEvent clusterChangedEvent, final Discovery.AckListener ackListener) {
    if (!master) {
        throw new IllegalStateException("Shouldn't publish state when not master");
    }
    LocalDiscovery[] members = members();
    if (members.length > 0) {
        Set<DiscoveryNode> nodesToPublishTo = new HashSet<>(members.length);
        for (LocalDiscovery localDiscovery : members) {
            if (localDiscovery.master) {
                continue;
            }
            nodesToPublishTo.add(localDiscovery.localNode());
        }
        publish(members, clusterChangedEvent, new AckClusterStatePublishResponseHandler(nodesToPublishTo, ackListener));
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
public void setReallocation(final ClusterService clusterService, final RoutingService routingService) {
    this.routingService = routingService;
    clusterService.add(new ClusterStateListener() {
        @Override
        public void clusterChanged(ClusterChangedEvent event) {
            boolean cleanCache = false;
            DiscoveryNode localNode = event.state().nodes().localNode();
            if (localNode != null) {
                if (localNode.masterNode() == true && event.localNodeMaster() == false) {
                    cleanCache = true;
                }
            } else {
                cleanCache = true;
            }
            if (cleanCache) {
                Releasables.close(asyncFetchStarted.values());
                asyncFetchStarted.clear();
                Releasables.close(asyncFetchStore.values());
                asyncFetchStore.clear();
            }
        }
    });
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
private boolean waitingShardsStartedOrUnassigned(ClusterChangedEvent event) {
    SnapshotsInProgress curr = event.state().custom(SnapshotsInProgress.TYPE);
    if (curr != null) {
        for (SnapshotsInProgress.Entry entry : curr.entries()) {
            if (entry.state() == State.STARTED && !entry.waitingIndices().isEmpty()) {
                for (String index : entry.waitingIndices().keySet()) {
                    if (event.indexRoutingTableChanged(index)) {
                        IndexRoutingTable indexShardRoutingTable = event.state().getRoutingTable().index(index);
                        for (ShardId shardId : entry.waitingIndices().get(index)) {
                            ShardRouting shardRouting = indexShardRoutingTable.shard(shardId.id()).primaryShard();
                            if (shardRouting != null && (shardRouting.started() || shardRouting.unassigned())) {
                                return true;
                            }
                        }
                    }
                }
            }
        }
    }
    return false;
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
private boolean removedNodesCleanupNeeded(ClusterChangedEvent event) {
    // Check if we just became the master
    boolean newMaster = !event.previousState().nodes().localNodeMaster();
    SnapshotsInProgress snapshotsInProgress = event.state().custom(SnapshotsInProgress.TYPE);
    if (snapshotsInProgress == null) {
        return false;
    }
    for (SnapshotsInProgress.Entry snapshot : snapshotsInProgress.entries()) {
        if (newMaster && (snapshot.state() == State.SUCCESS || snapshot.state() == State.INIT)) {
            // We just replaced old master and snapshots in intermediate states needs to be cleaned
            return true;
        }
        for (DiscoveryNode node : event.nodesDelta().removedNodes()) {
            for (ShardSnapshotStatus shardStatus : snapshot.shards().values()) {
                if (!shardStatus.state().completed() && node.getId().equals(shardStatus.nodeId())) {
                    // At least one shard was running on the removed node - we need to fail it
                    return true;
                }
            }
        }
    }
    return false;
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
@Override
public void clusterChanged(ClusterChangedEvent event) {
    try {
        SnapshotsInProgress prev = event.previousState().custom(SnapshotsInProgress.TYPE);
        SnapshotsInProgress curr = event.state().custom(SnapshotsInProgress.TYPE);

        if (prev == null) {
            if (curr != null) {
                processIndexShardSnapshots(event);
            }
        } else if (prev.equals(curr) == false) {
                processIndexShardSnapshots(event);
        }
        String masterNodeId = event.state().nodes().masterNodeId();
        if (masterNodeId != null && masterNodeId.equals(event.previousState().nodes().masterNodeId()) == false) {
            syncShardStatsOnNewMaster(event);
        }

    } catch (Throwable t) {
        logger.warn("Failed to update snapshot state ", t);
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
@Override
public void clusterChanged(ClusterChangedEvent event) {
    if(has5xNodes == null || event.nodesChanged()) {
        has5xNodes = Boolean.valueOf(clusterHas5xNodes(event.state()));
        if(log.isTraceEnabled()) {
            log.trace("has5xNodes: {}", has5xNodes);
        }
    }
    
    final List<String> indicesCreated = event.indicesCreated();
    final List<Index> indicesDeleted = event.indicesDeleted();
    if(has5xIndices == null || !indicesCreated.isEmpty() || !indicesDeleted.isEmpty()) {
        has5xIndices = Boolean.valueOf(clusterHas5xIndices(event.state()));
        if(log.isTraceEnabled()) {
            log.trace("has5xIndices: {}", has5xIndices);
        }
    }
    
    if(nodes == null || event.nodesChanged()) {
        nodes = event.state().nodes();
        if(log.isDebugEnabled()) {
            log.debug("Cluster Info Holder now initialized for 'nodes'");
        }
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
@Override
public synchronized void applyClusterState(final ClusterChangedEvent event) {
    if (!lifecycle.started()) {
        return;
    }

    final ClusterState state = event.state();

    // we need to clean the shards and indices we have on this node, since we
    // are going to recover them again once state persistence is disabled (no master / not recovered)
    // TODO: feels hacky, a block disables state persistence, and then we clean the allocated shards, maybe another flag in blocks?
    if (state.blocks().disableStatePersistence()) {
        for (AllocatedIndex<? extends Shard> indexService : indicesService) {
            indicesService.removeIndex(indexService.index(), NO_LONGER_ASSIGNED,
                "cleaning index (disabled block persistence)"); // also cleans shards
        }
        return;
    }

    updateFailedShardsCache(state);

    deleteIndices(event); // also deletes shards of deleted indices

    removeUnallocatedIndices(event); // also removes shards of removed indices

    failMissingShards(state);

    removeShards(state);   // removes any local shards that doesn't match what the master expects

    updateIndices(event); // can also fail shards, but these are then guaranteed to be in failedShardsCache

    createIndices(state);

    createOrUpdateShards(state);
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
/**
 * Removes indices that have no shards allocated to this node. This does not delete the shard data as we wait for enough
 * shard copies to exist in the cluster before deleting shard data (triggered by {@link org.elasticsearch.indices.store.IndicesStore}).
 *
 * @param event the cluster changed event
 */
private void removeUnallocatedIndices(final ClusterChangedEvent event) {
    final ClusterState state = event.state();
    final String localNodeId = state.nodes().getLocalNodeId();
    assert localNodeId != null;

    Set<Index> indicesWithShards = new HashSet<>();
    RoutingNode localRoutingNode = state.getRoutingNodes().node(localNodeId);
    if (localRoutingNode != null) { // null e.g. if we are not a data node
        for (ShardRouting shardRouting : localRoutingNode) {
            indicesWithShards.add(shardRouting.index());
        }
    }

    for (AllocatedIndex<? extends Shard> indexService : indicesService) {
        Index index = indexService.index();
        if (indicesWithShards.contains(index) == false) {
            // if the cluster change indicates a brand new cluster, we only want
            // to remove the in-memory structures for the index and not delete the
            // contents on disk because the index will later be re-imported as a
            // dangling index
            final IndexMetaData indexMetaData = state.metaData().index(index);
            assert indexMetaData != null || event.isNewCluster() :
                "index " + index + " does not exist in the cluster state, it should either " +
                    "have been deleted or the cluster must be new";
            final AllocatedIndices.IndexRemovalReason reason =
                indexMetaData != null && indexMetaData.getState() == IndexMetaData.State.CLOSE ? CLOSED : NO_LONGER_ASSIGNED;
            logger.debug("{} removing index, [{}]", index, reason);
            indicesService.removeIndex(index, reason, "removing index (no shards allocated)");
        }
    }
}
 

import org.elasticsearch.cluster.ClusterChangedEvent; //导入依赖的package包/类
private void updateIndices(ClusterChangedEvent event) {
    if (!event.metaDataChanged()) {
        return;
    }
    final ClusterState state = event.state();
    for (AllocatedIndex<? extends Shard> indexService : indicesService) {
        final Index index = indexService.index();
        final IndexMetaData currentIndexMetaData = indexService.getIndexSettings().getIndexMetaData();
        final IndexMetaData newIndexMetaData = state.metaData().index(index);
        assert newIndexMetaData != null : "index " + index + " should have been removed by deleteIndices";
        if (ClusterChangedEvent.indexMetaDataChanged(currentIndexMetaData, newIndexMetaData)) {
            indexService.updateMetaData(newIndexMetaData);
            try {
                if (indexService.updateMapping(newIndexMetaData) && sendRefreshMapping) {
                    nodeMappingRefreshAction.nodeMappingRefresh(state.nodes().getMasterNode(),
                        new NodeMappingRefreshAction.NodeMappingRefreshRequest(newIndexMetaData.getIndex().getName(),
                            newIndexMetaData.getIndexUUID(), state.nodes().getLocalNodeId())
                    );
                }
            } catch (Exception e) {
                indicesService.removeIndex(indexService.index(), FAILURE, "removing index (mapping update failed)");

                // fail shards that would be created or updated by createOrUpdateShards
                RoutingNode localRoutingNode = state.getRoutingNodes().node(state.nodes().getLocalNodeId());
                if (localRoutingNode != null) {
                    for (final ShardRouting shardRouting : localRoutingNode) {
                        if (shardRouting.index().equals(index) && failedShardsCache.containsKey(shardRouting.shardId()) == false) {
                            sendFailShard(shardRouting, "failed to update mapping for index", e, state);
                        }
                    }
                }
            }
        }
    }
}