core/src/main/java/org/elasticsearch/cluster/service/ClusterService.java
public <T> void submitStateUpdateTasks(final String source,
                                           final Map<T, ClusterStateTaskListener> tasks, final ClusterStateTaskConfig config,
                                           final ClusterStateTaskExecutor<T> executor) {
        if (!lifecycle.started()) {
            return;
        }
        if (tasks.isEmpty()) {
            return;
        }
        try {
            // convert to an identity map to check for dups based on update tasks semantics of using identity instead of equal
            final IdentityHashMap<T, ClusterStateTaskListener> tasksIdentity = new IdentityHashMap<>(tasks);
            final List<UpdateTask<T>> updateTasks = tasksIdentity.entrySet().stream().map(
                entry -> new UpdateTask<>(source, entry.getKey(), config, executor, safe(entry.getValue(), logger))
            ).collect(Collectors.toList());

            synchronized (updateTasksPerExecutor) {
                LinkedHashSet<UpdateTask> existingTasks = updateTasksPerExecutor.computeIfAbsent(executor,
                    k -> new LinkedHashSet<>(updateTasks.size()));
                for (@SuppressWarnings("unchecked") UpdateTask<T> existing : existingTasks) {
                    if (tasksIdentity.containsKey(existing.task)) {
                        throw new IllegalStateException("task [" + executor.describeTasks(Collections.singletonList(existing.task)) +
                            "] with source [" + source + "] is already queued");
                    }
                }
                existingTasks.addAll(updateTasks);
            }

            final UpdateTask<T> firstTask = updateTasks.get(0);

            final TimeValue timeout = config.timeout();
            if (timeout != null) {
                updateTasksExecutor.execute(firstTask, threadPool.scheduler(), timeout, () -> threadPool.generic().execute(() -> {
                    final ArrayList<UpdateTask<T>> toRemove = new ArrayList<>();
                    for (UpdateTask<T> task : updateTasks) {
                        if (task.processed.getAndSet(true) == false) {
                            logger.debug("cluster state update task [{}] timed out after [{}]", source, timeout);
                            toRemove.add(task);
                        }
                    }
                    if (toRemove.isEmpty() == false) {
                        ClusterStateTaskExecutor<T> clusterStateTaskExecutor = toRemove.get(0).executor;
                        synchronized (updateTasksPerExecutor) {
                            LinkedHashSet<UpdateTask> existingTasks = updateTasksPerExecutor.get(clusterStateTaskExecutor);
                            if (existingTasks != null) {
                                existingTasks.removeAll(toRemove);
                                if (existingTasks.isEmpty()) {
                                    updateTasksPerExecutor.remove(clusterStateTaskExecutor);
                                }
                            }
                        }
                        for (UpdateTask<T> task : toRemove) {
                            task.listener.onFailure(source, new ProcessClusterEventTimeoutException(timeout, source));
                        }
                    }
                }));
            } else {
                updateTasksExecutor.execute(firstTask);
            }
        } catch (EsRejectedExecutionException e) {
            // ignore cases where we are shutting down..., there is really nothing interesting
            // to be done here...
            if (!lifecycle.stoppedOrClosed()) {
                throw e;
            }
        }
    }
/**
     * Submits a batch of cluster state update tasks; submitted updates are guaranteed to be processed together,
     * potentially with more tasks of the same executor.
     *
     * @param source   the source of the cluster state update task
     * @param tasks    a map of update tasks and their corresponding listeners
     * @param config   the cluster state update task configuration
     * @param executor the cluster state update task executor; tasks
     *                 that share the same executor will be executed
     *                 batches on this executor
     * @param <T>      the type of the cluster state update task state
     *
     */
Submits a batch of cluster state update tasks; submitted updates are guaranteed to be processed together,
potentially with more tasks of the same executor.


Submits a batch of cluster state update tasks ; submitted updates are guaranteed to be processed together potentially with more tasks of the same executor .
public <T> void submitStateUpdateTasks(final String source,
                                           final Map<T, ClusterStateTaskListener> tasks, final ClusterStateTaskConfig config,
                                           final ClusterStateTaskExecutor<T> executor) {
        if (!lifecycle.started()) {
            return;
        }
        if (tasks.isEmpty()) {
            return;
        }
        try {
            final IdentityHashMap<T, ClusterStateTaskListener> tasksIdentity = new IdentityHashMap<>(tasks);
            final List<UpdateTask<T>> updateTasks = tasksIdentity.entrySet().stream().map(
                entry -> new UpdateTask<>(source, entry.getKey(), config, executor, safe(entry.getValue(), logger))
            ).collect(Collectors.toList());
            synchronized (updateTasksPerExecutor) {
                LinkedHashSet<UpdateTask> existingTasks = updateTasksPerExecutor.computeIfAbsent(executor,
                    k -> new LinkedHashSet<>(updateTasks.size()));
                for (@SuppressWarnings("unchecked") UpdateTask<T> existing : existingTasks) {
                    if (tasksIdentity.containsKey(existing.task)) {
                        throw new IllegalStateException("task [" + executor.describeTasks(Collections.singletonList(existing.task)) +
                            "] with source [" + source + "] is already queued");
                    }
                }
                existingTasks.addAll(updateTasks);
            }
            final UpdateTask<T> firstTask = updateTasks.get(0);
            final TimeValue timeout = config.timeout();
            if (timeout != null) {
                updateTasksExecutor.execute(firstTask, threadPool.scheduler(), timeout, () -> threadPool.generic().execute(() -> {
                    final ArrayList<UpdateTask<T>> toRemove = new ArrayList<>();
                    for (UpdateTask<T> task : updateTasks) {
                        if (task.processed.getAndSet(true) == false) {
                            logger.debug("cluster state update task [{}] timed out after [{}]", source, timeout);
                            toRemove.add(task);
                        }
                    }
                    if (toRemove.isEmpty() == false) {
                        ClusterStateTaskExecutor<T> clusterStateTaskExecutor = toRemove.get(0).executor;
                        synchronized (updateTasksPerExecutor) {
                            LinkedHashSet<UpdateTask> existingTasks = updateTasksPerExecutor.get(clusterStateTaskExecutor);
                            if (existingTasks != null) {
                                existingTasks.removeAll(toRemove);
                                if (existingTasks.isEmpty()) {
                                    updateTasksPerExecutor.remove(clusterStateTaskExecutor);
                                }
                            }
                        }
                        for (UpdateTask<T> task : toRemove) {
                            task.listener.onFailure(source, new ProcessClusterEventTimeoutException(timeout, source));
                        }
                    }
                }));
            } else {
                updateTasksExecutor.execute(firstTask);
            }
        } catch (EsRejectedExecutionException e) {
            if (!lifecycle.stoppedOrClosed()) {
                throw e;
            }
        }
    }
public < T > void submitStateUpdateTasks ( final String source , final Map < T , ClusterStateTaskListener > tasks , final ClusterStateTaskConfig config , final ClusterStateTaskExecutor < T > executor ) { if ( ! lifecycle . started ( ) ) { return ; } if ( tasks . isEmpty ( ) ) { return ; } try { final IdentityHashMap < T , ClusterStateTaskListener > tasksIdentity = new IdentityHashMap < > ( tasks ) ; final List < UpdateTask < T > > updateTasks = tasksIdentity . entrySet ( ) . stream ( ) . map ( entry -> new UpdateTask < > ( source , entry . getKey ( ) , config , executor , safe ( entry . getValue ( ) , logger ) ) ) . collect ( Collectors . toList ( ) ) ; synchronized ( updateTasksPerExecutor ) { LinkedHashSet < UpdateTask > existingTasks = updateTasksPerExecutor . computeIfAbsent ( executor , k -> new LinkedHashSet < > ( updateTasks . size ( ) ) ) ; for ( @ SuppressWarnings ( "unchecked" ) UpdateTask < T > existing : existingTasks ) { if ( tasksIdentity . containsKey ( existing . task ) ) { throw new IllegalStateException ( "task [" + executor . describeTasks ( Collections . singletonList ( existing . task ) ) + "] with source [" + source + "] is already queued" ) ; } } existingTasks . addAll ( updateTasks ) ; } final UpdateTask < T > firstTask = updateTasks . get ( 0 ) ; final TimeValue timeout = config . timeout ( ) ; if ( timeout != null ) { updateTasksExecutor . execute ( firstTask , threadPool . scheduler ( ) , timeout , ( ) -> threadPool . generic ( ) . execute ( ( ) -> { final ArrayList < UpdateTask < T > > toRemove = new ArrayList < > ( ) ; for ( UpdateTask < T > task : updateTasks ) { if ( task . processed . getAndSet ( true ) == false ) { logger . debug ( "cluster state update task [{}] timed out after [{}]" , source , timeout ) ; toRemove . add ( task ) ; } } if ( toRemove . isEmpty ( ) == false ) { ClusterStateTaskExecutor < T > clusterStateTaskExecutor = toRemove . get ( 0 ) . executor ; synchronized ( updateTasksPerExecutor ) { LinkedHashSet < UpdateTask > existingTasks = updateTasksPerExecutor . get ( clusterStateTaskExecutor ) ; if ( existingTasks != null ) { existingTasks . removeAll ( toRemove ) ; if ( existingTasks . isEmpty ( ) ) { updateTasksPerExecutor . remove ( clusterStateTaskExecutor ) ; } } } for ( UpdateTask < T > task : toRemove ) { task . listener . onFailure ( source , new ProcessClusterEventTimeoutException ( timeout , source ) ) ; } } } ) ) ; } else { updateTasksExecutor . execute ( firstTask ) ; } } catch ( EsRejectedExecutionException e ) { if ( ! lifecycle . stoppedOrClosed ( ) ) { throw e ; } } }
Submits a batch of cluster state update tasks; submitted updates are guaranteed to be processed together,
 potentially with more tasks of the same executor.
Submits a batch of cluster state update tasks ; submitted updates are guaranteed to be processed together potentially with more tasks of the same executor .
// convert to an identity map to check for dups based on update tasks semantics of using identity instead of equal // ignore cases where we are shutting down..., there is really nothing interesting // to be done here...