@Test
public void testRejoinGroup() {

    EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
    EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
    PowerMock.replayAll();
    client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    // join the group once
    client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()));
    client.prepareResponse(syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.<String>emptyList(),
            Collections.singletonList(taskId0), Errors.NONE.code()));
    coordinator.ensureActiveGroup();
    assertEquals(0, rebalanceListener.revokedCount);
    assertEquals(1, rebalanceListener.assignedCount);
    assertFalse(rebalanceListener.assignment.failed());
    assertEquals(1L, rebalanceListener.assignment.offset());
    assertEquals(Collections.emptyList(), rebalanceListener.assignment.connectors());
    assertEquals(Collections.singletonList(taskId0), rebalanceListener.assignment.tasks());
    // and join the group again
    coordinator.requestRejoin();
    client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()));
    client.prepareResponse(syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.singletonList(connectorId),
            Collections.<ConnectorTaskId>emptyList(), Errors.NONE.code()));
    coordinator.ensureActiveGroup();
    assertEquals(1, rebalanceListener.revokedCount);
    assertEquals(Collections.emptyList(), rebalanceListener.revokedConnectors);
    assertEquals(Collections.singletonList(taskId0), rebalanceListener.revokedTasks);
    assertEquals(2, rebalanceListener.assignedCount);
    assertFalse(rebalanceListener.assignment.failed());
    assertEquals(1L, rebalanceListener.assignment.offset());
    assertEquals(Collections.singletonList(connectorId), rebalanceListener.assignment.connectors());
    assertEquals(Collections.emptyList(), rebalanceListener.assignment.tasks());
    PowerMock.verifyAll();
}