@Override
    public Map<String, ByteBuffer> performAssignment(String leaderId, String protocol,
                                                     List<JoinGroupResponseMember> allMemberMetadata,
                                                     WorkerCoordinator coordinator) {
        log.debug("Performing task assignment");

        Map<String, ExtendedWorkerState> memberConfigs = new HashMap<>();
        for (JoinGroupResponseMember member : allMemberMetadata) {
            memberConfigs.put(
                    member.memberId(),
                    IncrementalCooperativeConnectProtocol.deserializeMetadata(ByteBuffer.wrap(member.metadata())));
        }
        log.debug("Member configs: {}", memberConfigs);

        // The new config offset is the maximum seen by any member. We always perform assignment using this offset,
        // even if some members have fallen behind. The config offset used to generate the assignment is included in
        // the response so members that have fallen behind will not use the assignment until they have caught up.
        long maxOffset = memberConfigs.values().stream().map(ExtendedWorkerState::offset).max(Long::compare).get();
        log.debug("Max config offset root: {}, local snapshot config offsets root: {}",
                  maxOffset, coordinator.configSnapshot().offset());

        short protocolVersion = ConnectProtocolCompatibility.fromProtocol(protocol).protocolVersion();

        Long leaderOffset = ensureLeaderConfig(maxOffset, coordinator);
        if (leaderOffset == null) {
            Map<String, ExtendedAssignment> assignments = fillAssignments(
                    memberConfigs.keySet(), Assignment.CONFIG_MISMATCH,
                    leaderId, memberConfigs.get(leaderId).url(), maxOffset,
                    ClusterAssignment.EMPTY, 0, protocolVersion);
            return serializeAssignments(assignments, protocolVersion);
        }
        return performTaskAssignment(leaderId, leaderOffset, memberConfigs, coordinator, protocolVersion);
    }
@Test
    public void testProtocolV2() {
        // Sanity test to make sure that the right protocol is chosen during the assignment
        connectors.clear();
        String leader = "followMe";
        List<JoinGroupResponseData.JoinGroupResponseMember> memberMetadata = new ArrayList<>();
        ExtendedAssignment leaderAssignment = new ExtendedAssignment(
                IncrementalCooperativeConnectProtocol.CONNECT_PROTOCOL_V2,
                ConnectProtocol.Assignment.NO_ERROR,
                leader,
                "followMe:618",
                CONFIG_OFFSET,
                Collections.emptySet(),
                Collections.emptySet(),
                Collections.emptySet(),
                Collections.emptySet(),
                0
        );
        ExtendedWorkerState leaderState = new ExtendedWorkerState("followMe:618", CONFIG_OFFSET, leaderAssignment);
        JoinGroupResponseData.JoinGroupResponseMember leaderMetadata = new JoinGroupResponseData.JoinGroupResponseMember()
                .setMemberId(leader)
                .setMetadata(IncrementalCooperativeConnectProtocol.serializeMetadata(leaderState, true).array());
        memberMetadata.add(leaderMetadata);
        WorkerCoordinator coordinator = mock(WorkerCoordinator.class);
        when(coordinator.configSnapshot()).thenReturn(configState());
        Map<String, ByteBuffer> serializedAssignments = assignor.performAssignment(
                leader,
                ConnectProtocolCompatibility.SESSIONED.protocol(),
                memberMetadata,
                coordinator
        );
        serializedAssignments.forEach((worker, serializedAssignment) -> {
            ExtendedAssignment assignment = IncrementalCooperativeConnectProtocol.deserializeAssignment(serializedAssignment);
            assertEquals(
                    IncrementalCooperativeConnectProtocol.CONNECT_PROTOCOL_V2,
                    assignment.version(),
                    "Incorrect protocol version in assignment for worker " + worker
            );
        });
    }