import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
private void notifyNodeFailure(final DiscoveryNode node, final String reason) {
    try {
        threadPool.generic().execute(new Runnable() {
            @Override
            public void run() {
                for (Listener listener : listeners) {
                    listener.onNodeFailure(node, reason);
                }
            }
        });
    } catch (EsRejectedExecutionException ex) {
        logger.trace(
            (Supplier<?>) () -> new ParameterizedMessage(
                "[node  ] [{}] ignoring node failure (reason [{}]). Local node is shutting down",
                node,
                reason),
            ex);
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
@Override
public void onNodeDisconnected(final DiscoveryNode node) {
    try {
        threadPool.generic().execute( () -> {
            for (final TransportConnectionListener connectionListener : connectionListeners) {
                connectionListener.onNodeDisconnected(node);
            }
        });
        for (Map.Entry<Long, RequestHolder> entry : clientHandlers.entrySet()) {
            RequestHolder holder = entry.getValue();
            if (holder.node().equals(node)) {
                final RequestHolder holderToNotify = clientHandlers.remove(entry.getKey());
                if (holderToNotify != null) {
                    // callback that an exception happened, but on a different thread since we don't
                    // want handlers to worry about stack overflows
                    threadPool.generic().execute(() -> holderToNotify.handler().handleException(new NodeDisconnectedException(node,
                        holderToNotify.action())));
                }
            }
        }
    } catch (EsRejectedExecutionException ex) {
        logger.debug("Rejected execution on NodeDisconnected", ex);
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
@Override
public void onMaster() {
    this.isMaster = true;
    if (logger.isTraceEnabled()) {
        logger.trace("I have been elected master, scheduling a ClusterInfoUpdateJob");
    }
    try {
        // Submit a job that will start after DEFAULT_STARTING_INTERVAL, and reschedule itself after running
        threadPool.schedule(updateFrequency, executorName(), new SubmitReschedulingClusterInfoUpdatedJob());
        if (clusterService.state().getNodes().getDataNodes().size() > 1) {
            // Submit an info update job to be run immediately
            threadPool.executor(executorName()).execute(() -> maybeRefresh());
        }
    } catch (EsRejectedExecutionException ex) {
        if (logger.isDebugEnabled()) {
            logger.debug("Couldn't schedule cluster info update task - node might be shutting down", ex);
        }
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
public Discovery.AckListener createAckListener(ThreadPool threadPool, ClusterState newClusterState) {
    ArrayList<Discovery.AckListener> ackListeners = new ArrayList<>();

    //timeout straightaway, otherwise we could wait forever as the timeout thread has not started
    nonFailedTasks.stream().filter(task -> task.listener instanceof AckedClusterStateTaskListener).forEach(task -> {
        final AckedClusterStateTaskListener ackedListener = (AckedClusterStateTaskListener) task.listener;
        if (ackedListener.ackTimeout() == null || ackedListener.ackTimeout().millis() == 0) {
            ackedListener.onAckTimeout();
        } else {
            try {
                ackListeners.add(new AckCountDownListener(ackedListener, newClusterState.version(), newClusterState.nodes(),
                    threadPool));
            } catch (EsRejectedExecutionException ex) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Couldn't schedule timeout thread - node might be shutting down", ex);
                }
                //timeout straightaway, otherwise we could wait forever as the timeout thread has not started
                ackedListener.onAckTimeout();
            }
        }
    });

    return new DelegetingAckListener(ackListeners);
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
public AbstractAsyncBulkByScrollAction(WorkingBulkByScrollTask task, Logger logger, ParentTaskAssigningClient client,
        ThreadPool threadPool, Request mainRequest, ScriptService scriptService, ClusterState clusterState,
        ActionListener<BulkByScrollResponse> listener) {
    this.task = task;
    this.logger = logger;
    this.client = client;
    this.threadPool = threadPool;
    this.scriptService = scriptService;
    this.clusterState = clusterState;
    this.mainRequest = mainRequest;
    this.listener = listener;
    BackoffPolicy backoffPolicy = buildBackoffPolicy();
    bulkRetry = Retry.on(EsRejectedExecutionException.class).policy(BackoffPolicy.wrap(backoffPolicy, task::countBulkRetry));
    scrollSource = buildScrollableResultSource(backoffPolicy);
    scriptApplier = Objects.requireNonNull(buildScriptApplier(), "script applier must not be null");
    /*
     * Default to sorting by doc. We can't do this in the request itself because it is normal to *add* to the sorts rather than replace
     * them and if we add _doc as the first sort by default then sorts will never work.... So we add it here, only if there isn't
     * another sort.
     */
    List<SortBuilder<?>> sorts = mainRequest.getSearchRequest().source().sorts();
    if (sorts == null || sorts.isEmpty()) {
        mainRequest.getSearchRequest().source().sort(fieldSort("_doc"));
    }
    mainRequest.getSearchRequest().source().version(needsSourceDocumentVersions());
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
public void testOnRejectionCausesCancellation() throws Exception {
    final TimeValue delay = TimeValue.timeValueMillis(10L);
    terminate(threadPool);
    threadPool = new ThreadPool(Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), "fixed delay tests").build()) {
        @Override
        public ScheduledFuture<?> schedule(TimeValue delay, String executor, Runnable command) {
            if (command instanceof ReschedulingRunnable) {
                ((ReschedulingRunnable) command).onRejection(new EsRejectedExecutionException());
            } else {
                fail("this should only be called with a rescheduling runnable in this test");
            }
            return null;
        }
    };
    Runnable runnable = () -> {};
    ReschedulingRunnable reschedulingRunnable = new ReschedulingRunnable(runnable, delay, Names.GENERIC, threadPool);
    assertTrue(reschedulingRunnable.isCancelled());
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
public void testAsyncRetryFailsAfterBacksOff() throws Exception {
    BackoffPolicy backoff = BackoffPolicy.constantBackoff(DELAY, CALLS_TO_FAIL - 1);
    AssertingListener listener = new AssertingListener();

    BulkRequest bulkRequest = createBulkRequest();
    Retry.on(EsRejectedExecutionException.class)
            .policy(backoff)
            .withAsyncBackoff(bulkClient, bulkRequest, listener);

    listener.awaitCallbacksCalled();

    listener.assertOnResponseCalled();
    listener.assertResponseWithFailures();
    listener.assertResponseWithNumberOfItems(bulkRequest.numberOfActions());
    listener.assertOnFailureNeverCalled();
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
/**
 * Mimicks a ThreadPool rejecting execution of the task.
 */
public void testThreadPoolRejectionsAbortRequest() throws Exception {
    testTask.rethrottle(1);
    setupClient(new TestThreadPool(getTestName()) {
        @Override
        public ScheduledFuture<?> schedule(TimeValue delay, String name, Runnable command) {
            // While we're here we can check that the sleep made it through
            assertThat(delay.nanos(), greaterThan(0L));
            assertThat(delay.seconds(), lessThanOrEqualTo(10L));
            ((AbstractRunnable) command).onRejection(new EsRejectedExecutionException("test"));
            return null;
        }
    });
    ScrollableHitSource.Response response = new ScrollableHitSource.Response(false, emptyList(), 0, emptyList(), null);
    simulateScrollResponse(new DummyAsyncBulkByScrollAction(), timeValueNanos(System.nanoTime()), 10, response);
    ExecutionException e = expectThrows(ExecutionException.class, () -> listener.get());
    assertThat(e.getMessage(), equalTo("EsRejectedExecutionException[test]"));
    assertThat(client.scrollsCleared, contains(scrollId));

    // When the task is rejected we don't increment the throttled timer
    assertEquals(timeValueMillis(0), testTask.getStatus().getThrottled());
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
@Override
public void run() {
    if (closed) {
        return;
    }
    if (keysToClean.isEmpty()) {
        schedule();
        return;
    }
    try {
        threadPool.executor(ThreadPool.Names.GENERIC).execute(new Runnable() {
            @Override
            public void run() {
                reap();
                schedule();
            }
        });
    } catch (EsRejectedExecutionException ex) {
        logger.debug("Can not run ReaderCleaner - execution rejected", ex);
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
@Override
public void onMaster() {
    this.isMaster = true;
    if (logger.isTraceEnabled()) {
        logger.trace("I have been elected master, scheduling a ClusterInfoUpdateJob");
    }
    try {
        // Submit a job that will start after DEFAULT_STARTING_INTERVAL, and reschedule itself after running
        threadPool.schedule(updateFrequency, executorName(), new SubmitReschedulingClusterInfoUpdatedJob());
        if (clusterService.state().getNodes().getDataNodes().size() > 1) {
            // Submit an info update job to be run immediately
            threadPool.executor(executorName()).execute(new Runnable() {
                @Override
                public void run() {
                    maybeRefresh();
                }
            });
        }
    } catch (EsRejectedExecutionException ex) {
        if (logger.isDebugEnabled()) {
            logger.debug("Couldn't schedule cluster info update task - node might be shutting down", ex);
        }
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
@Override
public void execute(BulkRequest bulkRequest, long executionId) {
    boolean afterCalled = false;
    try {
        listener.beforeBulk(executionId, bulkRequest);
        BulkResponse bulkResponse = Retry
                .on(EsRejectedExecutionException.class)
                .policy(backoffPolicy)
                .withSyncBackoff(client, bulkRequest);
        afterCalled = true;
        listener.afterBulk(executionId, bulkRequest, bulkResponse);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        logger.info("Bulk request {} has been cancelled.", e, executionId);
        if (!afterCalled) {
            listener.afterBulk(executionId, bulkRequest, e);
        }
    } catch (Throwable t) {
        logger.warn("Failed to execute bulk request {}.", t, executionId);
        if (!afterCalled) {
            listener.afterBulk(executionId, bulkRequest, t);
        }
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
private void schedule() {
    boolean success = false;
    while (!success) {
        if (closed) {
            break;
        }
        try {
            threadPool.schedule(cleanInterval, ThreadPool.Names.SAME,
                    this);
            success = true;
        } catch (final EsRejectedExecutionException ex) {
            logger.warn("Can not schedule Reaper - execution rejected",
                    ex);
            try {
                Thread.sleep(1000);
            } catch (final InterruptedException e) {
                // ignore
            }
        }
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
@Override
public void insertDiscovery(List<Token> tokens) throws IOException {
    batchHistogram.update(tokens.size());
    if (tokens.size() == 0) return;

    Timer.Context ctx = writeTimer.time();
    try {

        BulkRequestBuilder bulk = client.prepareBulk();

        for (Token token : tokens) {
            bulk.add(createSingleRequest(token));
        }

        bulk.execute().actionGet();
    } catch (EsRejectedExecutionException esEx) {
        log.error(("Error during bulk insert to ES with status: [" + esEx.status() + "] " +
                "with message: [" + esEx.getDetailedMessage() + "]"));
        throw esEx;
    } finally {
        ctx.stop();
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    super.setUp();
    createIndex("source");
    // Build the test data. Don't use indexRandom because that won't work consistently with such small thread pools.
    BulkRequestBuilder bulk = client().prepareBulk();
    for (int i = 0; i < DOC_COUNT; i++) {
        bulk.add(client().prepareIndex("source", "test").setSource("foo", "bar " + i));
    }
    Retry retry = Retry.on(EsRejectedExecutionException.class).policy(BackoffPolicy.exponentialBackoff());
    BulkResponse response = retry.withSyncBackoff(client(), bulk.request());
    assertFalse(response.buildFailureMessage(), response.hasFailures());
    client().admin().indices().prepareRefresh("source").get();
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
public void testParseRejection() throws Exception {
    // The rejection comes through in the handler because the mocked http response isn't marked as an error
    AtomicBoolean called = new AtomicBoolean();
    // Handling a scroll rejection is the same as handling a search rejection so we reuse the verification code
    Consumer<Response> checkResponse = r -> {
        assertFalse(r.isTimedOut());
        assertEquals(FAKE_SCROLL_ID, r.getScrollId());
        assertEquals(4, r.getTotalHits());
        assertThat(r.getFailures(), hasSize(1));
        assertEquals("test", r.getFailures().get(0).getIndex());
        assertEquals((Integer) 0, r.getFailures().get(0).getShardId());
        assertEquals("87A7NvevQxSrEwMbtRCecg", r.getFailures().get(0).getNodeId());
        assertThat(r.getFailures().get(0).getReason(), instanceOf(EsRejectedExecutionException.class));
        assertEquals("rejected execution of [email protected] on "
                + "EsThreadPoolExecutor[search, queue capacity = 1000, org.elasticsearch.common.util.concurrent."
                + "[email protected][Running, pool size = 7, active threads = 7, queued tasks = 1000, "
                + "completed tasks = 4182]]", r.getFailures().get(0).getReason().getMessage());
        assertThat(r.getHits(), hasSize(1));
        assertEquals("test", r.getHits().get(0).getIndex());
        assertEquals("test", r.getHits().get(0).getType());
        assertEquals("AVToMiC250DjIiBO3yJ_", r.getHits().get(0).getId());
        assertEquals("{\"test\":\"test1\"}", r.getHits().get(0).getSource().utf8ToString());
        called.set(true);
    };
    sourceWithMockedRemoteCall("rejection.json").doStart(checkResponse);
    assertTrue(called.get());
    called.set(false);
    sourceWithMockedRemoteCall("rejection.json").doStartNextScroll("scroll", timeValueMillis(0), checkResponse);
    assertTrue(called.get());
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
private void notifyMasterFailure(final DiscoveryNode masterNode, final Throwable cause, final String reason) {
    if (notifiedMasterFailure.compareAndSet(false, true)) {
        try {
            threadPool.generic().execute(() -> {
                for (Listener listener : listeners) {
                    listener.onMasterFailure(masterNode, cause, reason);
                }
            });
        } catch (EsRejectedExecutionException e) {
            logger.error("master failure notification was rejected, it's highly likely the node is shutting down", e);
        }
        stop("master failure, " + reason);
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
@Override
protected void onAfterInLifecycle() {
    try {
        threadPool.schedule(pingSchedule, ThreadPool.Names.GENERIC, this);
    } catch (EsRejectedExecutionException ex) {
        if (ex.isExecutorShutdown()) {
            logger.debug("couldn't schedule new ping execution, executor is shutting down", ex);
        } else {
            throw ex;
        }
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
/**
 * Adds a cluster state listener that is expected to be removed during a short period of time.
 * If provided, the listener will be notified once a specific time has elapsed.
 *
 * NOTE: the listener is not removed on timeout. This is the responsibility of the caller.
 */
public void addTimeoutListener(@Nullable final TimeValue timeout, final TimeoutClusterStateListener listener) {
    if (lifecycle.stoppedOrClosed()) {
        listener.onClose();
        return;
    }
    // call the post added notification on the same event thread
    try {
        threadPoolExecutor.execute(new SourcePrioritizedRunnable(Priority.HIGH, "_add_listener_") {
            @Override
            public void run() {
                if (timeout != null) {
                    NotifyTimeout notifyTimeout = new NotifyTimeout(listener, timeout);
                    notifyTimeout.future = threadPool.schedule(timeout, ThreadPool.Names.GENERIC, notifyTimeout);
                    onGoingTimeouts.add(notifyTimeout);
                }
                timeoutClusterStateListeners.add(listener);
                listener.postAdded();
            }
        });
    } catch (EsRejectedExecutionException e) {
        if (lifecycle.stoppedOrClosed()) {
            listener.onClose();
        } else {
            throw e;
        }
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
@Override
public void onAfter() {
    // if this has not been cancelled reschedule it to run again
    if (run) {
        try {
            threadPool.schedule(interval, executor, this);
        } catch (final EsRejectedExecutionException e) {
            onRejection(e);
        }
    }
}
 

import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; //导入依赖的package包/类
public void testSyncRetryBacksOff() throws Exception {
    BackoffPolicy backoff = BackoffPolicy.constantBackoff(DELAY, CALLS_TO_FAIL);

    BulkRequest bulkRequest = createBulkRequest();
    BulkResponse response = Retry
            .on(EsRejectedExecutionException.class)
            .policy(backoff)
            .withSyncBackoff(bulkClient, bulkRequest);

    assertFalse(response.hasFailures());
    assertThat(response.getItems().length, equalTo(bulkRequest.numberOfActions()));
}