import org.elasticsearch.common.Priority; //导入依赖的package包/类
protected void ensureStableCluster(int nodeCount, TimeValue timeValue, boolean local, @Nullable String viaNode) {
    if (viaNode == null) {
        viaNode = randomFrom(internalCluster().getNodeNames());
    }
    logger.debug("ensuring cluster is stable with [{}] nodes. access node: [{}]. timeout: [{}]", nodeCount, viaNode, timeValue);
    ClusterHealthResponse clusterHealthResponse = client(viaNode).admin().cluster().prepareHealth()
        .setWaitForEvents(Priority.LANGUID)
        .setWaitForNodes(Integer.toString(nodeCount))
        .setTimeout(timeValue)
        .setLocal(local)
        .setWaitForNoRelocatingShards(true)
        .get();
    if (clusterHealthResponse.isTimedOut()) {
        ClusterStateResponse stateResponse = client(viaNode).admin().cluster().prepareState().get();
        fail("failed to reach a stable cluster of [" + nodeCount + "] nodes. Tried via [" + viaNode + "]. last cluster state:\n"
            + stateResponse.getState());
    }
    assertThat(clusterHealthResponse.isTimedOut(), is(false));
    ensureFullyConnectedCluster();
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
public void deleteIndices(final DeleteIndexClusterStateUpdateRequest request,
        final ActionListener<ClusterStateUpdateResponse> listener) {
    if (request.indices() == null || request.indices().length == 0) {
        throw new IllegalArgumentException("Index name is required");
    }

    clusterService.submitStateUpdateTask("delete-index " + Arrays.toString(request.indices()),
        new AckedClusterStateUpdateTask<ClusterStateUpdateResponse>(Priority.URGENT, request, listener) {

        @Override
        protected ClusterStateUpdateResponse newResponse(boolean acknowledged) {
            return new ClusterStateUpdateResponse(acknowledged);
        }

        @Override
        public ClusterState execute(final ClusterState currentState) {
            return deleteIndices(currentState, Sets.newHashSet(request.indices()));
        }
    });
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
@Override
protected Runnable wrapRunnable(Runnable command) {
    if (command instanceof PrioritizedRunnable) {
        if ((command instanceof TieBreakingPrioritizedRunnable)) {
            return command;
        }
        Priority priority = ((PrioritizedRunnable) command).priority();
        return new TieBreakingPrioritizedRunnable(super.wrapRunnable(command), priority, insertionOrder.incrementAndGet());
    } else if (command instanceof PrioritizedFutureTask) {
        return command;
    } else { // it might be a callable wrapper...
        if (command instanceof TieBreakingPrioritizedRunnable) {
            return command;
        }
        return new TieBreakingPrioritizedRunnable(super.wrapRunnable(command), Priority.NORMAL, insertionOrder.incrementAndGet());
    }
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    int size = in.readVInt();
    if (size == 0) {
        indices = Strings.EMPTY_ARRAY;
    } else {
        indices = new String[size];
        for (int i = 0; i < indices.length; i++) {
            indices[i] = in.readString();
        }
    }
    timeout = new TimeValue(in);
    if (in.readBoolean()) {
        waitForStatus = ClusterHealthStatus.fromValue(in.readByte());
    }
    waitForNoRelocatingShards = in.readBoolean();
    waitForActiveShards = ActiveShardCount.readFrom(in);
    waitForNodes = in.readString();
    if (in.readBoolean()) {
        waitForEvents = Priority.readFrom(in);
    }
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
public void testUpdateMappingWithoutTypeMultiObjects() throws Exception {
    client().admin().indices().prepareCreate("test")
            .setSettings(
                    Settings.builder()
                            .put("index.number_of_shards", 1)
                            .put("index.number_of_replicas", 0)
            ).execute().actionGet();
    client().admin().cluster().prepareHealth().setWaitForEvents(Priority.LANGUID).setWaitForGreenStatus().execute().actionGet();

    PutMappingResponse putMappingResponse = client().admin().indices().preparePutMapping("test").setType("doc")
            .setSource("{\"properties\":{\"date\":{\"type\":\"integer\"}}}", XContentType.JSON)
            .execute().actionGet();

    assertThat(putMappingResponse.isAcknowledged(), equalTo(true));

    GetMappingsResponse getMappingsResponse = client().admin().indices().prepareGetMappings("test").execute().actionGet();
    assertThat(getMappingsResponse.mappings().get("test").get("doc").source().toString(),
            equalTo("{\"doc\":{\"properties\":{\"date\":{\"type\":\"integer\"}}}}"));
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
public void testUpdateMappingWithConflicts() throws Exception {
    client().admin().indices().prepareCreate("test")
            .setSettings(
                    Settings.builder()
                            .put("index.number_of_shards", 2)
                            .put("index.number_of_replicas", 0)
            ).addMapping("type", "{\"type\":{\"properties\":{\"body\":{\"type\":\"text\"}}}}", XContentType.JSON)
            .execute().actionGet();
    client().admin().cluster().prepareHealth().setWaitForEvents(Priority.LANGUID).setWaitForGreenStatus().execute().actionGet();

    try {
        client().admin().indices().preparePutMapping("test").setType("type")
                .setSource("{\"type\":{\"properties\":{\"body\":{\"type\":\"integer\"}}}}", XContentType.JSON).execute().actionGet();
        fail("Expected MergeMappingException");
    } catch (IllegalArgumentException e) {
        assertThat(e.getMessage(), containsString("mapper [body] of different type, current_type [text], merged_type [integer]"));
    }
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
public void testUpdateMappingNoChanges() throws Exception {
    client().admin().indices().prepareCreate("test")
            .setSettings(
                    Settings.builder()
                            .put("index.number_of_shards", 2)
                            .put("index.number_of_replicas", 0)
            ).addMapping("type", "{\"type\":{\"properties\":{\"body\":{\"type\":\"text\"}}}}", XContentType.JSON)
            .execute().actionGet();
    client().admin().cluster().prepareHealth().setWaitForEvents(Priority.LANGUID).setWaitForGreenStatus().execute().actionGet();

    PutMappingResponse putMappingResponse = client().admin().indices().preparePutMapping("test").setType("type")
            .setSource("{\"type\":{\"properties\":{\"body\":{\"type\":\"text\"}}}}", XContentType.JSON)
            .execute().actionGet();

    //no changes, we return
    assertThat(putMappingResponse.isAcknowledged(), equalTo(true));
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
public void execute(Runnable command, final ScheduledExecutorService timer, final TimeValue timeout, final Runnable timeoutCallback) {
    if (command instanceof PrioritizedRunnable) {
        command = new TieBreakingPrioritizedRunnable((PrioritizedRunnable) command, insertionOrder.incrementAndGet());
    } else if (!(command instanceof PrioritizedFutureTask)) { // it might be a callable wrapper...
        command = new TieBreakingPrioritizedRunnable(command, Priority.NORMAL, insertionOrder.incrementAndGet());
    }
    super.execute(command);
    if (timeout.nanos() >= 0) {
        if (command instanceof TieBreakingPrioritizedRunnable) {
            ((TieBreakingPrioritizedRunnable) command).scheduleTimeout(timer, timeoutCallback, timeout);
        } else {
            // We really shouldn't be here. The only way we can get here if somebody created PrioritizedFutureTask
            // and passed it to execute, which doesn't make much sense
            throw new UnsupportedOperationException("Execute with timeout is not supported for future tasks");
        }
    }
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    int size = in.readVInt();
    if (size == 0) {
        indices = Strings.EMPTY_ARRAY;
    } else {
        indices = new String[size];
        for (int i = 0; i < indices.length; i++) {
            indices[i] = in.readString();
        }
    }
    timeout = readTimeValue(in);
    if (in.readBoolean()) {
        waitForStatus = ClusterHealthStatus.fromValue(in.readByte());
    }
    waitForRelocatingShards = in.readInt();
    waitForActiveShards = in.readInt();
    waitForNodes = in.readString();
    if (in.readBoolean()) {
        waitForEvents = Priority.readFrom(in);
    }
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
public void testJustMasterNode() throws Exception {
    logger.info("--> cleaning nodes");

    logger.info("--> starting 1 master node non data");
    internalCluster().startNode(Settings.builder().put(Node.NODE_DATA_SETTING.getKey(), false).build());

    logger.info("--> create an index");
    client().admin().indices().prepareCreate("test").setWaitForActiveShards(ActiveShardCount.NONE).execute().actionGet();

    logger.info("--> closing master node");
    internalCluster().closeNonSharedNodes(false);

    logger.info("--> starting 1 master node non data again");
    internalCluster().startNode(Settings.builder().put(Node.NODE_DATA_SETTING.getKey(), false).build());

    logger.info("--> waiting for test index to be created");
    ClusterHealthResponse health = client().admin().cluster().prepareHealth().setWaitForEvents(Priority.LANGUID).setIndices("test")
        .execute().actionGet();
    assertThat(health.isTimedOut(), equalTo(false));

    logger.info("--> verify we have an index");
    ClusterStateResponse clusterStateResponse = client().admin().cluster().prepareState().setIndices("test").execute().actionGet();
    assertThat(clusterStateResponse.getState().metaData().hasIndex("test"), equalTo(true));
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
/**
 * Note, this test can only work as long as we have a single thread executor executing the state update tasks!
 */
public void testPrioritizedTasks() throws Exception {
    BlockingTask block = new BlockingTask(Priority.IMMEDIATE);
    clusterService.submitStateUpdateTask("test", block);
    int taskCount = randomIntBetween(5, 20);

    // will hold all the tasks in the order in which they were executed
    List<PrioritizedTask> tasks = new ArrayList<>(taskCount);
    CountDownLatch latch = new CountDownLatch(taskCount);
    for (int i = 0; i < taskCount; i++) {
        Priority priority = randomFrom(Priority.values());
        clusterService.submitStateUpdateTask("test", new PrioritizedTask(priority, latch, tasks));
    }

    block.close();
    latch.await();

    Priority prevPriority = null;
    for (PrioritizedTask task : tasks) {
        if (prevPriority == null) {
            prevPriority = task.priority();
        } else {
            assertThat(task.priority().sameOrAfter(prevPriority), is(true));
        }
    }
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
public void testPriorityQueue() throws Exception {
    PriorityBlockingQueue<Priority> queue = new PriorityBlockingQueue<>();
    List<Priority> priorities = Arrays.asList(Priority.values());
    Collections.shuffle(priorities, random());

    for (Priority priority : priorities) {
        queue.add(priority);
    }

    Priority prevPriority = null;
    while (!queue.isEmpty()) {
        if (prevPriority == null) {
            prevPriority = queue.poll();
        } else {
            assertThat(queue.poll().after(prevPriority), is(true));
        }
    }
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
private ClusterHealthStatus ensureColor(ClusterHealthStatus clusterHealthStatus, TimeValue timeout, String... indices) {
    String color = clusterHealthStatus.name().toLowerCase(Locale.ROOT);
    String method = "ensure" + Strings.capitalize(color);

    ClusterHealthRequest healthRequest = Requests.clusterHealthRequest(indices)
        .timeout(timeout)
        .waitForStatus(clusterHealthStatus)
        .waitForEvents(Priority.LANGUID)
        .waitForNoRelocatingShards(true)
        // We currently often use ensureGreen or ensureYellow to check whether the cluster is back in a good state after shutting down
        // a node. If the node that is stopped is the master node, another node will become master and publish a cluster state where it
        // is master but where the node that was stopped hasn't been removed yet from the cluster state. It will only subsequently
        // publish a second state where the old master is removed. If the ensureGreen/ensureYellow is timed just right, it will get to
        // execute before the second cluster state update removes the old master and the condition ensureGreen / ensureYellow will
        // trivially hold if it held before the node was shut down. The following "waitForNodes" condition ensures that the node has
        // been removed by the master so that the health check applies to the set of nodes we expect to be part of the cluster.
        .waitForNodes(Integer.toString(cluster().size()));

    ClusterHealthResponse actionGet = client().admin().cluster().health(healthRequest).actionGet();
    if (actionGet.isTimedOut()) {
        logger.info("{} timed out, cluster state:\n{}\n{}",
            method,
            client().admin().cluster().prepareState().get().getState(),
            client().admin().cluster().preparePendingClusterTasks().get());
        fail("timed out waiting for " + color + " state");
    }
    assertThat("Expected at least " + clusterHealthStatus + " but got " + actionGet.getStatus(),
        actionGet.getStatus().value(), lessThanOrEqualTo(clusterHealthStatus.value()));
    logger.debug("indices {} are {}", indices.length == 0 ? "[_all]" : indices, color);
    return actionGet.getStatus();
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
/**
 * Ensures the cluster has a green state via the cluster health API. This method will also wait for relocations.
 * It is useful to ensure that all action on the cluster have finished and all shards that were currently relocating
 * are now allocated and started.
 *
 * @param timeout time out value to set on {@link org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest}
 */
public ClusterHealthStatus ensureGreen(TimeValue timeout, String... indices) {
    ClusterHealthResponse actionGet = client().admin().cluster()
            .health(Requests.clusterHealthRequest(indices).timeout(timeout).waitForGreenStatus().waitForEvents(Priority.LANGUID)
                    .waitForNoRelocatingShards(true)).actionGet();
    if (actionGet.isTimedOut()) {
        logger.info("ensureGreen timed out, cluster state:\n{}\n{}", client().admin().cluster().prepareState().get().getState(),
            client().admin().cluster().preparePendingClusterTasks().get());
        assertThat("timed out waiting for green state", actionGet.isTimedOut(), equalTo(false));
    }
    assertThat(actionGet.getStatus(), equalTo(ClusterHealthStatus.GREEN));
    logger.debug("indices {} are green", indices.length == 0 ? "[_all]" : indices);
    return actionGet.getStatus();
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
/**
 * processes or queues an incoming join request.
 * <p>
 * Note: doesn't do any validation. This should have been done before.
 */
public synchronized void handleJoinRequest(final DiscoveryNode node, final MembershipAction.JoinCallback callback) {
    if (electionContext != null) {
        electionContext.addIncomingJoin(node, callback);
        checkPendingJoinsAndElectIfNeeded();
    } else {
        clusterService.submitStateUpdateTask("zen-disco-node-join",
            node, ClusterStateTaskConfig.build(Priority.URGENT),
            joinTaskExecutor, new JoinTaskListener(callback, logger));
    }
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
public synchronized void closeAndProcessPending(String reason) {
    innerClose();
    Map<DiscoveryNode, ClusterStateTaskListener> tasks = getPendingAsTasks();
    final String source = "zen-disco-election-stop [" + reason + "]";
    tasks.put(FINISH_ELECTION_TASK, electionFinishedListener);
    clusterService.submitStateUpdateTasks(source, tasks, ClusterStateTaskConfig.build(Priority.URGENT), joinTaskExecutor);
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
private void handleMasterGone(final DiscoveryNode masterNode, final Throwable cause, final String reason) {
    if (lifecycleState() != Lifecycle.State.STARTED) {
        // not started, ignore a master failure
        return;
    }
    if (localNodeMaster()) {
        // we might get this on both a master telling us shutting down, and then the disconnect failure
        return;
    }

    logger.info((Supplier<?>) () -> new ParameterizedMessage("master_left [{}], reason [{}]", masterNode, reason), cause);

    clusterService.submitStateUpdateTask("master_failed (" + masterNode + ")", new LocalClusterUpdateTask(Priority.IMMEDIATE) {

        @Override
        public ClusterTasksResult<LocalClusterUpdateTask> execute(ClusterState currentState) {
            if (!masterNode.equals(currentState.nodes().getMasterNode())) {
                // master got switched on us, no need to send anything
                return unchanged();
            }

            // flush any pending cluster states from old master, so it will not be set as master again
            publishClusterState.pendingStatesQueue().failAllStatesAndClear(new ElasticsearchException("master left [{}]", reason));

            return rejoin(currentState, "master left (reason = " + reason + ")");
        }

        @Override
        public void onFailure(String source, Exception e) {
            logger.error((Supplier<?>) () -> new ParameterizedMessage("unexpected failure during [{}]", source), e);
        }

    });
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
@Override
public void writeTo(StreamOutput out) throws IOException {
    super.writeTo(out);
    if (indices == null) {
        out.writeVInt(0);
    } else {
        out.writeVInt(indices.length);
        for (String index : indices) {
            out.writeString(index);
        }
    }
    timeout.writeTo(out);
    if (waitForStatus == null) {
        out.writeBoolean(false);
    } else {
        out.writeBoolean(true);
        out.writeByte(waitForStatus.value());
    }
    out.writeInt(waitForRelocatingShards);
    out.writeInt(waitForActiveShards);
    out.writeString(waitForNodes);
    if (waitForEvents == null) {
        out.writeBoolean(false);
    } else {
        out.writeBoolean(true);
        Priority.writeTo(waitForEvents, out);
    }
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    ClusterHealthRequest clusterHealthRequest = clusterHealthRequest(Strings.splitStringByCommaToArray(request.param("index")));
    clusterHealthRequest.local(request.paramAsBoolean("local", clusterHealthRequest.local()));
    clusterHealthRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterHealthRequest.masterNodeTimeout()));
    clusterHealthRequest.timeout(request.paramAsTime("timeout", clusterHealthRequest.timeout()));
    String waitForStatus = request.param("wait_for_status");
    if (waitForStatus != null) {
        clusterHealthRequest.waitForStatus(ClusterHealthStatus.valueOf(waitForStatus.toUpperCase(Locale.ROOT)));
    }
    clusterHealthRequest.waitForNoRelocatingShards(
            request.paramAsBoolean("wait_for_no_relocating_shards", clusterHealthRequest.waitForNoRelocatingShards()));
    if (request.hasParam("wait_for_relocating_shards")) {
        // wait_for_relocating_shards has been removed in favor of wait_for_no_relocating_shards
        throw new IllegalArgumentException("wait_for_relocating_shards has been removed, " +
                                           "use wait_for_no_relocating_shards [true/false] instead");
    }
    String waitForActiveShards = request.param("wait_for_active_shards");
    if (waitForActiveShards != null) {
        clusterHealthRequest.waitForActiveShards(ActiveShardCount.parseString(waitForActiveShards));
    }
    clusterHealthRequest.waitForNodes(request.param("wait_for_nodes", clusterHealthRequest.waitForNodes()));
    if (request.param("wait_for_events") != null) {
        clusterHealthRequest.waitForEvents(Priority.valueOf(request.param("wait_for_events").toUpperCase(Locale.ROOT)));
    }
    return channel -> client.admin().cluster().health(clusterHealthRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.common.Priority; //导入依赖的package包/类
@Override
public void clusterChanged(final ClusterChangedEvent event) {
    logger.debug("[{}] received cluster event, [{}]", tribeName, event.source());
    clusterService.submitStateUpdateTask(
            "cluster event from " + tribeName,
            event,
            ClusterStateTaskConfig.build(Priority.NORMAL),
            executor,
            (source, e) -> logger.warn((Supplier<?>) () -> new ParameterizedMessage("failed to process [{}]", source), e));
}