import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的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.ClusterStateListener; //导入依赖的package包/类
public MockNode(DiscoveryNode discoveryNode, MockTransportService service,
                @Nullable ClusterStateListener listener, Logger logger) {
    this.discoveryNode = discoveryNode;
    this.service = service;
    this.listener = listener;
    this.logger = logger;
    this.clusterState = ClusterState.builder(CLUSTER_NAME).nodes(DiscoveryNodes.builder()
        .add(discoveryNode).localNodeId(discoveryNode.getId()).build()).build();
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
@Inject
public DecommissioningService(Settings settings,
                              final ClusterService clusterService,
                              NodeSettingsService nodeSettingsService,
                              TransportSQLAction sqlAction,
                              TransportSQLBulkAction sqlBulkAction,
                              final TransportClusterHealthAction healthAction,
                              final TransportClusterUpdateSettingsAction updateSettingsAction) {
    super(settings);
    this.clusterService = clusterService;
    this.sqlAction = sqlAction;
    this.sqlBulkAction = sqlBulkAction;
    this.healthAction = healthAction;
    this.updateSettingsAction = updateSettingsAction;

    ApplySettings applySettings = new ApplySettings();
    applySettings.onRefreshSettings(settings);
    nodeSettingsService.addListener(applySettings);

    clusterService.add(new ClusterStateListener() {
        @Override
        public void clusterChanged(ClusterChangedEvent event) {
            removeRemovedNodes(event);
        }
    });
    try {
        Signal signal = new Signal("USR2");
        Signal.handle(signal, this);
    } catch (IllegalArgumentException e) {
        logger.warn("SIGUSR2 signal not supported on {}.", System.getProperty("os.name"), e);
    }
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
@Override
public void remove(ClusterStateListener listener) {
    clusterStateListeners.remove(listener);
    priorityClusterStateListeners.remove(listener);
    lastClusterStateListeners.remove(listener);
    postAppliedListeners.remove(listener);
    for (Iterator<NotifyTimeout> it = onGoingTimeouts.iterator(); it.hasNext(); ) {
        NotifyTimeout timeout = it.next();
        if (timeout.listener.equals(listener)) {
            timeout.cancel();
            it.remove();
        }
    }
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
@Override
protected void doStart() throws ElasticsearchException {
    logger.info("Starting EventHookService");

    clusterService.add((ClusterStateListener) this);
    clusterService.add((LocalNodeMasterListener) this);
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
@Override
protected void doStop() throws ElasticsearchException {
    logger.info("Stopping EventHookService");

    clusterService.remove((LocalNodeMasterListener) this);
    clusterService.remove((ClusterStateListener) this);
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
/**
 * Add a listener for updated cluster states
 */
public void addListener(ClusterStateListener listener) {
    clusterStateListeners.add(listener);
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
/**
 * Removes a listener for updated cluster states.
 */
public void removeListener(ClusterStateListener listener) {
    clusterStateListeners.remove(listener);
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
public MockNode createMockNode(String name, final Settings basSettings, @Nullable ClusterStateListener listener) throws Exception {
    return createMockNode(name, basSettings, listener, threadPool, logger, nodes);
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
public void testSerializationFailureDuringDiffPublishing() throws Exception {
    MockNode nodeA = createMockNode("nodeA", Settings.EMPTY, new ClusterStateListener() {
        @Override
        public void clusterChanged(ClusterChangedEvent event) {
            fail("Shouldn't send cluster state to myself");
        }
    }).setAsMaster();

    MockNode nodeB = createMockNode("nodeB");

    // Initial cluster state with both states - the second node still shouldn't get
    // diff even though it's present in the previous cluster state
    DiscoveryNodes discoveryNodes = DiscoveryNodes.builder(nodeA.nodes()).add(nodeB.discoveryNode).build();
    ClusterState previousClusterState = ClusterState.builder(CLUSTER_NAME).nodes(discoveryNodes).build();
    ClusterState clusterState = ClusterState.builder(previousClusterState).incrementVersion().build();
    publishStateAndWait(nodeA.action, clusterState, previousClusterState);
    assertSameStateFromFull(nodeB.clusterState, clusterState);

    // cluster state update - add block
    previousClusterState = clusterState;
    clusterState = ClusterState.builder(clusterState).blocks(ClusterBlocks.builder()
        .addGlobalBlock(MetaData.CLUSTER_READ_ONLY_BLOCK)).incrementVersion().build();

    ClusterState unserializableClusterState = new ClusterState(clusterState.version(), clusterState.stateUUID(), clusterState) {
        @Override
        public Diff<ClusterState> diff(ClusterState previousState) {
            return new Diff<ClusterState>() {
                @Override
                public ClusterState apply(ClusterState part) {
                    fail("this diff shouldn't be applied");
                    return part;
                }

                @Override
                public void writeTo(StreamOutput out) throws IOException {
                    throw new IOException("Simulated failure of diff serialization");
                }
            };
        }
    };
    try {
        publishStateAndWait(nodeA.action, unserializableClusterState, previousClusterState);
        fail("cluster state published despite of diff errors");
    } catch (Discovery.FailedToCommitClusterStateException e) {
        assertThat(e.getCause(), notNullValue());
        assertThat(e.getCause().getMessage(), containsString("failed to serialize"));
    }
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
@Override
public void addFirst(ClusterStateListener listener) {
    priorityClusterStateListeners.add(listener);
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
@Override
public void addLast(ClusterStateListener listener) {
    lastClusterStateListeners.add(listener);
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
@Override
public void add(ClusterStateListener listener) {
    clusterStateListeners.add(listener);
}
 

import org.elasticsearch.cluster.ClusterStateListener; //导入依赖的package包/类
@Override
protected void doStart() throws ElasticsearchException {
    logger.info("doStart()");

    poller = new Thread(new Runnable() {

        @Override
        public void run() {
            while (true) {
                try {
                    final IndexRequest req = queue.take();
                    if (req == null) {
                        break;
                    }
                    bulk.add(req);
                } catch (final InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            logger.info("poller thread died");
        }
    });
    poller.setDaemon(true);
    poller.setName("poller");
    poller.start();

    this.clusterService.add(new ClusterStateListener() {

        @Override
        public void clusterChanged(final ClusterChangedEvent event) {

            if (event.localNodeMaster()) {
                isMaster = true;
            } else {
                isMaster = false;
            }

        }
    });

    this.indicesService.indicesLifecycle().addListener(auditIndicesLsListener);

}