@Test
public void testJoinLeaderCannotAssign() {

    // If the selected leader can't get up to the maximum offset, it will fail to assign and we should immediately
    // need to retry the join.
    // When the first round fails, we'll take an updated config snapshot
    EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
    EasyMock.expect(configStorage.snapshot()).andReturn(configState2);
    PowerMock.replayAll();
    final String memberId = "member";
    client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    // config mismatch results in assignment error
    client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE.code()));
    MockClient.RequestMatcher matcher = new MockClient.RequestMatcher() {
        @Override
        public boolean matches(ClientRequest request) {
            SyncGroupRequest sync = new SyncGroupRequest(request.request().body());
            return sync.memberId().equals(memberId) &&
                    sync.generationId() == 1 &&
                    sync.groupAssignment().isEmpty();
        }
    };
    client.prepareResponse(matcher, syncGroupResponse(ConnectProtocol.Assignment.CONFIG_MISMATCH, "leader", 10L,
            Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList(), Errors.NONE.code()));
    client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE.code()));
    client.prepareResponse(matcher, syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader", 1L,
            Collections.<String>emptyList(), Collections.singletonList(taskId0), Errors.NONE.code()));
    coordinator.ensureActiveGroup();
    PowerMock.verifyAll();
}