streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
boolean tryToCompleteRestoration() {
        boolean allRunning = true;

        final List<Task> restoringTasks = new LinkedList<>();
        for (final Task task : tasks.values()) {
            if (task.state() == CREATED) {
                try {
                    task.initializeIfNeeded();
                } catch (final LockException | TimeoutException e) {
                    // it is possible that if there are multiple threads within the instance that one thread
                    // trying to grab the task from the other, while the other has not released the lock since
                    // it did not participate in the rebalance. In this case we can just retry in the next iteration
                    log.debug("Could not initialize {} due to the following exception; will retry", task.id(), e);
                    allRunning = false;
                }
            }

            if (task.state() == RESTORING) {
                restoringTasks.add(task);
            }
        }

        if (allRunning && !restoringTasks.isEmpty()) {
            final Set<TopicPartition> restored = changelogReader.completedChangelogs();
            for (final Task task : restoringTasks) {
                if (restored.containsAll(task.changelogPartitions())) {
                    try {
                        task.completeRestoration();
                    } catch (final TimeoutException e) {
                        log.debug("Could not complete restoration for {} due to {}; will retry", task.id(), e);

                        allRunning = false;
                    }
                } else {
                    // we found a restoring task that isn't done restoring, which is evidence that
                    // not all tasks are running
                    allRunning = false;
                }
            }
        }

        if (allRunning) {
            // we can call resume multiple times since it is idempotent.
            mainConsumer.resume(mainConsumer.assignment());
        }

        return allRunning;
    }
/**
     * Tries to initialize any new or still-uninitialized tasks, then checks if they can/have completed restoration.
     *
     * @throws IllegalStateException If store gets registered after initialized is already finished
     * @throws StreamsException if the store's change log does not contain the partition
     * @return {@code true} if all tasks are fully restored
     */
Tries to initialize any new or still-uninitialized tasks, then checks if they can/have completed restoration.
@throws IllegalStateException If store gets registered after initialized is already finished
@throws StreamsException if the store's change log does not contain the partition
@return  true if all tasks are fully restored
Tries to initialize any new or still - uninitialized tasks then checks if they can / have completed restoration . @throws IllegalStateException If store gets registered after initialized is already finished @throws StreamsException if the store ' s change log does not contain the partition @return true if all tasks are fully restored
boolean tryToCompleteRestoration() {
        boolean allRunning = true;
        final List<Task> restoringTasks = new LinkedList<>();
        for (final Task task : tasks.values()) {
            if (task.state() == CREATED) {
                try {
                    task.initializeIfNeeded();
                } catch (final LockException | TimeoutException e) {
                    log.debug("Could not initialize {} due to the following exception; will retry", task.id(), e);
                    allRunning = false;
                }
            }
            if (task.state() == RESTORING) {
                restoringTasks.add(task);
            }
        }
        if (allRunning && !restoringTasks.isEmpty()) {
            final Set<TopicPartition> restored = changelogReader.completedChangelogs();
            for (final Task task : restoringTasks) {
                if (restored.containsAll(task.changelogPartitions())) {
                    try {
                        task.completeRestoration();
                    } catch (final TimeoutException e) {
                        log.debug("Could not complete restoration for {} due to {}; will retry", task.id(), e);
                        allRunning = false;
                    }
                } else {
                    allRunning = false;
                }
            }
        }
        if (allRunning) {
            mainConsumer.resume(mainConsumer.assignment());
        }
        return allRunning;
    }
boolean tryToCompleteRestoration ( ) { boolean allRunning = true ; final List < Task > restoringTasks = new LinkedList < > ( ) ; for ( final Task task : tasks . values ( ) ) { if ( task . state ( ) == CREATED ) { try { task . initializeIfNeeded ( ) ; } catch ( final LockException | TimeoutException e ) { log . debug ( "Could not initialize {} due to the following exception; will retry" , task . id ( ) , e ) ; allRunning = false ; } } if ( task . state ( ) == RESTORING ) { restoringTasks . add ( task ) ; } } if ( allRunning && ! restoringTasks . isEmpty ( ) ) { final Set < TopicPartition > restored = changelogReader . completedChangelogs ( ) ; for ( final Task task : restoringTasks ) { if ( restored . containsAll ( task . changelogPartitions ( ) ) ) { try { task . completeRestoration ( ) ; } catch ( final TimeoutException e ) { log . debug ( "Could not complete restoration for {} due to {}; will retry" , task . id ( ) , e ) ; allRunning = false ; } } else { allRunning = false ; } } } if ( allRunning ) { mainConsumer . resume ( mainConsumer . assignment ( ) ) ; } return allRunning ; }
Tries to initialize any new or still-uninitialized tasks, then checks if they can/have completed restoration.
Tries to initialize any new or still - uninitialized tasks then checks if they can / have completed restoration .
// it is possible that if there are multiple threads within the instance that one thread // trying to grab the task from the other, while the other has not released the lock since // it did not participate in the rebalance. In this case we can just retry in the next iteration // we found a restoring task that isn't done restoring, which is evidence that // not all tasks are running // we can call resume multiple times since it is idempotent.