class IncrementalCooperativeAssignor implements ConnectAssignor {

    protected void handleLostAssignments(ConnectorsAndTasks lostAssignments,
                                         ConnectorsAndTasks newSubmissions,
                                         List<WorkerLoad> completeWorkerAssignment,
                                         Map<String, ExtendedWorkerState> memberConfigs) {
        if (lostAssignments.isEmpty()) {
            resetDelay();
            return;
        }

        final long now = time.milliseconds();
        log.debug("Found the following connectors and tasks missing from previous assignments: "
                + lostAssignments);

        if (scheduledRebalance <= 0 && memberConfigs.keySet().containsAll(previousMembers)) {
            log.debug("No worker seems to have departed the group during the rebalance. The "
                    + "missing assignments that the leader is detecting are probably due to some "
                    + "workers failing to receive the new assignments in the previous rebalance. "
                    + "Will reassign missing tasks as new tasks");
            newSubmissions.connectors().addAll(lostAssignments.connectors());
            newSubmissions.tasks().addAll(lostAssignments.tasks());
            return;
        }

        if (scheduledRebalance > 0 && now >= scheduledRebalance) {
            // delayed rebalance expired and it's time to assign resources
            log.debug("Delayed rebalance expired. Reassigning lost tasks");
            Optional<WorkerLoad> candidateWorkerLoad = Optional.empty();
            if (!candidateWorkersForReassignment.isEmpty()) {
                candidateWorkerLoad = pickCandidateWorkerForReassignment(completeWorkerAssignment);
            }

            if (candidateWorkerLoad.isPresent()) {
                WorkerLoad workerLoad = candidateWorkerLoad.get();
                log.debug("A candidate worker has been found to assign lost tasks: {}", workerLoad.worker());
                lostAssignments.connectors().forEach(workerLoad::assign);
                lostAssignments.tasks().forEach(workerLoad::assign);
            } else {
                log.debug("No single candidate worker was found to assign lost tasks. Treating lost tasks as new tasks");
                newSubmissions.connectors().addAll(lostAssignments.connectors());
                newSubmissions.tasks().addAll(lostAssignments.tasks());
            }
            resetDelay();
        } else {
            candidateWorkersForReassignment
                    .addAll(candidateWorkersForReassignment(completeWorkerAssignment));
            if (now < scheduledRebalance) {
                // a delayed rebalance is in progress, but it's not yet time to reassign
                // unaccounted resources
                delay = calculateDelay(now);
                log.debug("Delayed rebalance in progress. Task reassignment is postponed. New computed rebalance delay: {}", delay);
            } else {
                // This means scheduledRebalance == 0
                // We could also also extract the current minimum delay from the group, to make
                // independent of consecutive leader failures, but this optimization is skipped
                // at the moment
                delay = maxDelay;
                log.debug("Resetting rebalance delay to the max: {}. scheduledRebalance: {} now: {} diff scheduledRebalance - now: {}",
                        delay, scheduledRebalance, now, scheduledRebalance - now);
            }
            scheduledRebalance = now + delay;
        }
    }

    public  IncrementalCooperativeAssignor(LogContext logContext, Time time, int maxDelay);

}