@Test
public void testNormalJoinGroupLeader() {

    final String consumerId = "leader";
    subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener);
    subscriptions.needReassignment();
    // ensure metadata is up-to-date for leader
    metadata.setTopics(Arrays.asList(topicName));
    metadata.update(cluster, time.milliseconds());
    client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    // normal join group
    Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, Arrays.asList(topicName));
    partitionAssignor.prepare(Collections.singletonMap(consumerId, Arrays.asList(tp)));
    client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, 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().containsKey(consumerId);
        }
    }, syncGroupResponse(Arrays.asList(tp), Errors.NONE.code()));
    coordinator.ensurePartitionAssignment();
    assertFalse(subscriptions.partitionAssignmentNeeded());
    assertEquals(Collections.singleton(tp), subscriptions.assignedPartitions());
    assertEquals(1, rebalanceListener.revokedCount);
    assertEquals(Collections.emptySet(), rebalanceListener.revoked);
    assertEquals(1, rebalanceListener.assignedCount);
    assertEquals(Collections.singleton(tp), rebalanceListener.assigned);
}