protected void handleLostAssignments(ConnectorsAndTasks lostAssignments,
                                         ConnectorsAndTasks.Builder lostAssignmentsToReassign,
                                         List<WorkerLoad> completeWorkerAssignment) {
        // There are no lost assignments and there have been no successive revoking rebalances
        if (lostAssignments.isEmpty() && !revokedInPrevious) {
            resetDelay();
            return;
        }

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

        Set<String> activeMembers = completeWorkerAssignment.stream()
                .map(WorkerLoad::worker)
                .collect(Collectors.toSet());
        if (scheduledRebalance <= 0 && activeMembers.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");
            lostAssignmentsToReassign.addAll(lostAssignments);
            return;
        } else if (maxDelay == 0) {
            log.debug("Scheduled rebalance delays are disabled ({} = 0); "
                    + "reassigning all lost connectors and tasks immediately",
                    SCHEDULED_REBALANCE_MAX_DELAY_MS_CONFIG
            );
            lostAssignmentsToReassign.addAll(lostAssignments);
            return;
        }

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

            if (!candidateWorkerLoad.isEmpty()) {
                log.debug("Assigning lost tasks to {} candidate workers: {}", 
                        candidateWorkerLoad.size(),
                        candidateWorkerLoad.stream().map(WorkerLoad::worker).collect(Collectors.joining(",")));
                Iterator<WorkerLoad> candidateWorkerIterator = candidateWorkerLoad.iterator();
                for (String connector : lostAssignments.connectors()) {
                    // Loop over the candidate workers as many times as it takes
                    if (!candidateWorkerIterator.hasNext()) {
                        candidateWorkerIterator = candidateWorkerLoad.iterator();
                    }
                    WorkerLoad worker = candidateWorkerIterator.next();
                    log.debug("Assigning connector id {} to member {}", connector, worker.worker());
                    worker.assign(connector);
                }
                candidateWorkerIterator = candidateWorkerLoad.iterator();
                for (ConnectorTaskId task : lostAssignments.tasks()) {
                    if (!candidateWorkerIterator.hasNext()) {
                        candidateWorkerIterator = candidateWorkerLoad.iterator();
                    }
                    WorkerLoad worker = candidateWorkerIterator.next();
                    log.debug("Assigning task id {} to member {}", task, worker.worker());
                    worker.assign(task);
                }
            } else {
                log.debug("No single candidate worker was found to assign lost tasks. Treating lost tasks as new tasks");
                lostAssignmentsToReassign.addAll(lostAssignments);
            }
            resetDelay();
            // Resetting the flag as now we can permit successive revoking rebalances.
            // since we have gone through the full rebalance delay
            revokedInPrevious = false;
        } 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 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;
        }
    }
@Test
    public void testLostAssignmentHandlingWhenWorkerBouncesBackButFinallyLeaves() {
        // Customize assignor for this test case
        time = new MockTime();
        initAssignor();

        assertTrue(assignor.candidateWorkersForReassignment.isEmpty());
        assertEquals(0, assignor.scheduledRebalance);
        assertEquals(0, assignor.delay);

        Map<String, WorkerLoad> configuredAssignment = new HashMap<>();
        configuredAssignment.put("worker0", workerLoad("worker0", 0, 2, 0, 4));
        configuredAssignment.put("worker1", workerLoad("worker1", 2, 2, 4, 4));
        configuredAssignment.put("worker2", workerLoad("worker2", 4, 2, 8, 4));

        // No lost assignments
        assignor.handleLostAssignments(new ConnectorsAndTasks.Builder().build(),
                new ConnectorsAndTasks.Builder(),
                new ArrayList<>(configuredAssignment.values()));

        assertEquals(Collections.emptySet(),
            assignor.candidateWorkersForReassignment,
            "Wrong set of workers for reassignments");
        assertEquals(0, assignor.scheduledRebalance);
        assertEquals(0, assignor.delay);

        assignor.previousMembers = new HashSet<>(configuredAssignment.keySet());

        String veryFlakyWorker = "worker1";
        WorkerLoad lostLoad = configuredAssignment.remove(veryFlakyWorker);
        ConnectorsAndTasks lostAssignments = new ConnectorsAndTasks.Builder()
                .with(lostLoad.connectors(), lostLoad.tasks()).build();

        // Lost assignments detected - No candidate worker has appeared yet (worker with no assignments)
        assignor.handleLostAssignments(lostAssignments, new ConnectorsAndTasks.Builder(),
                new ArrayList<>(configuredAssignment.values()));

        assertEquals(Collections.emptySet(),
            assignor.candidateWorkersForReassignment,
            "Wrong set of workers for reassignments");
        assertEquals(time.milliseconds() + rebalanceDelay, assignor.scheduledRebalance);
        assertEquals(rebalanceDelay, assignor.delay);

        assignor.previousMembers = new HashSet<>(configuredAssignment.keySet());
        time.sleep(rebalanceDelay / 2);
        rebalanceDelay /= 2;

        // A new worker (probably returning worker) has joined
        configuredAssignment.put(veryFlakyWorker, new WorkerLoad.Builder(veryFlakyWorker).build());
        assignor.handleLostAssignments(lostAssignments, new ConnectorsAndTasks.Builder(),
                new ArrayList<>(configuredAssignment.values()));

        assertEquals(Collections.singleton(veryFlakyWorker),
            assignor.candidateWorkersForReassignment,
            "Wrong set of workers for reassignments");
        assertEquals(time.milliseconds() + rebalanceDelay, assignor.scheduledRebalance);
        assertEquals(rebalanceDelay, assignor.delay);

        assignor.previousMembers = new HashSet<>(configuredAssignment.keySet());
        time.sleep(rebalanceDelay);

        // The returning worker leaves permanently after joining briefly during the delay
        configuredAssignment.remove(veryFlakyWorker);
        ConnectorsAndTasks.Builder lostAssignmentsToReassign = new ConnectorsAndTasks.Builder();
        assignor.handleLostAssignments(lostAssignments, lostAssignmentsToReassign,
                new ArrayList<>(configuredAssignment.values()));

        assertTrue(lostAssignmentsToReassign.build().connectors().containsAll(lostAssignments.connectors()),
            "Wrong assignment of lost connectors");
        assertTrue(lostAssignmentsToReassign.build().tasks().containsAll(lostAssignments.tasks()),
            "Wrong assignment of lost tasks");
        assertEquals(Collections.emptySet(),
            assignor.candidateWorkersForReassignment,
            "Wrong set of workers for reassignments");
        assertEquals(0, assignor.scheduledRebalance);
        assertEquals(0, assignor.delay);
    }