@Test
public void testNormalJoinGroupFollower() {

    final String consumerId = "consumer";
    subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener);
    subscriptions.needReassignment();
    client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    // normal join group
    client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code()));
    client.prepareResponse(new MockClient.RequestMatcher() {
        @Override
        public boolean matches(ClientRequest request) {
            SyncGroupRequest sync = new SyncGroupRequest(request.request().body());
            return sync.memberId().equals(consumerId) &&
                    sync.generationId() == 1 &&
                    sync.groupAssignment().isEmpty();
        }
    }, syncGroupResponse(Arrays.asList(tp), Errors.NONE.code()));
    coordinator.ensurePartitionAssignment();
    assertFalse(subscriptions.partitionAssignmentNeeded());
    assertEquals(Collections.singleton(tp), subscriptions.assignedPartitions());
    assertEquals(1, rebalanceListener.revokedCount);
    assertEquals(1, rebalanceListener.assignedCount);
    assertEquals(Collections.singleton(tp), rebalanceListener.assigned);
}