protected Map<String, ByteBuffer> performTaskAssignment(String leaderId, long maxOffset,
                                                            Map<String, ExtendedWorkerState> memberConfigs,
                                                            WorkerCoordinator coordinator, short protocolVersion) {
        log.debug("Performing task assignment during generation: {} with memberId: {}",
                coordinator.generationId(), coordinator.memberId());
        Map<String, ConnectorsAndTasks> memberAssignments = transformValues(
                memberConfigs,
                memberConfig -> new ConnectorsAndTasks.Builder()
                        .with(memberConfig.assignment().connectors(), memberConfig.assignment().tasks())
                        .build()
        );
        ClusterAssignment clusterAssignment = performTaskAssignment(
                coordinator.configSnapshot(),
                coordinator.lastCompletedGenerationId(),
                coordinator.generationId(),
                memberAssignments
        );

        coordinator.leaderState(new LeaderState(memberConfigs, clusterAssignment.allAssignedConnectors(), clusterAssignment.allAssignedTasks()));

        Map<String, ExtendedAssignment> assignments =
                fillAssignments(memberConfigs.keySet(), Assignment.NO_ERROR, leaderId,
                        memberConfigs.get(leaderId).url(), maxOffset,
                        clusterAssignment,
                        delay, protocolVersion);

        log.debug("Actual assignments: {}", assignments);
        return serializeAssignments(assignments, protocolVersion);
    }
@Test
    public void testLeaderStateUpdated() {
        // Sanity test to make sure that the coordinator's leader state is actually updated after a rebalance
        connectors.clear();
        String leader = "followMe";
        Map<String, ExtendedWorkerState> workerStates = new HashMap<>();
        workerStates.put(leader, new ExtendedWorkerState("followMe:618", CONFIG_OFFSET, ExtendedAssignment.empty()));
        WorkerCoordinator coordinator = mock(WorkerCoordinator.class);
        when(coordinator.configSnapshot()).thenReturn(configState());
        assignor.performTaskAssignment(
                leader,
                CONFIG_OFFSET,
                workerStates,
                coordinator,
                IncrementalCooperativeConnectProtocol.CONNECT_PROTOCOL_V2
        );
        verify(coordinator).leaderState(notNull());
    }