ConsumerCoordinator extends AbstractCoordinator { public void ensurePartitionAssignment() { if (subscriptions.partitionsAutoAssigned()) { if (subscriptions.hasPatternSubscription()) { client.ensureFreshMetadata(); } ensureActiveGroup(); } } ConsumerCoordinator(ConsumerNetworkClient client,
        String groupId,
        int sessionTimeoutMs,
        int heartbeatIntervalMs,
        List<PartitionAssignor> assignors,
        Metadata metadata,
        SubscriptionState subscriptions,
        Metrics metrics,
        String metricGrpPrefix,
        Time time,
        long retryBackoffMs,
        OffsetCommitCallback defaultOffsetCommitCallback,
        boolean autoCommitEnabled,
        long autoCommitIntervalMs,
        ConsumerInterceptors<?, ?> interceptors,
        boolean excludeInternalTopics); @Override String protocolType(); @Override List<ProtocolMetadata> metadata(); @Override boolean needRejoin(); void refreshCommittedOffsetsIfNeeded(); Map<TopicPartition, OffsetAndMetadata> fetchCommittedOffsets(Set<TopicPartition> partitions); void ensurePartitionAssignment(); @Override void close(); void commitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback); void commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets);  }
@Test(expected = GroupAuthorizationException.class) public void testGroupReadUnauthorized() { subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupLeaderResponse(0, "memberId", Collections.<String, List<String>>emptyMap(), Errors.GROUP_AUTHORIZATION_FAILED.code())); coordinator.ensurePartitionAssignment(); }
@Test(expected = ApiException.class) public void testJoinGroupInvalidGroupId() { final String consumerId = "leader"; subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); metadata.setTopics(Arrays.asList(topicName)); metadata.update(cluster, time.milliseconds()); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupLeaderResponse(0, consumerId, Collections.<String, List<String>>emptyMap(), Errors.INVALID_GROUP_ID.code())); coordinator.ensurePartitionAssignment(); }
@Test public void testNormalJoinGroupLeader() { final String consumerId = "leader"; subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); metadata.setTopics(Arrays.asList(topicName)); metadata.update(cluster, time.milliseconds()); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); 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); }
@Test public void testWakeupDuringJoin() { final String consumerId = "leader"; subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); metadata.setTopics(Arrays.asList(topicName)); metadata.update(cluster, time.milliseconds()); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); 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())); consumerClient.wakeup(); try { coordinator.ensurePartitionAssignment(); } catch (WakeupException e) { } client.prepareResponse(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); }
@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(); 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); }
@Test public void testMaybeLeaveGroup() { final String consumerId = "consumer"; subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); final AtomicBoolean received = new AtomicBoolean(false); client.prepareResponse(new MockClient.RequestMatcher() { @Override public boolean matches(ClientRequest request) { received.set(true); LeaveGroupRequest leaveRequest = new LeaveGroupRequest(request.request().body()); return leaveRequest.memberId().equals(consumerId) && leaveRequest.groupId().equals(groupId); } }, new LeaveGroupResponse(Errors.NONE.code()).toStruct()); coordinator.maybeLeaveGroup(); assertTrue(received.get()); assertEquals(JoinGroupRequest.UNKNOWN_MEMBER_ID, coordinator.memberId); assertEquals(OffsetCommitRequest.DEFAULT_GENERATION_ID, coordinator.generation); }
@Test(expected = KafkaException.class) public void testUnexpectedErrorOnSyncGroup() { final String consumerId = "consumer"; subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Collections.<TopicPartition>emptyList(), Errors.UNKNOWN.code())); coordinator.ensurePartitionAssignment(); }
@Test public void testUnknownMemberIdOnSyncGroup() { final String consumerId = "consumer"; subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Collections.<TopicPartition>emptyList(), Errors.UNKNOWN_MEMBER_ID.code())); client.prepareResponse(new MockClient.RequestMatcher() { @Override public boolean matches(ClientRequest request) { JoinGroupRequest joinRequest = new JoinGroupRequest(request.request().body()); return joinRequest.memberId().equals(JoinGroupRequest.UNKNOWN_MEMBER_ID); } }, joinGroupFollowerResponse(2, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); assertFalse(subscriptions.partitionAssignmentNeeded()); assertEquals(Collections.singleton(tp), subscriptions.assignedPartitions()); }
@Test public void testRebalanceInProgressOnSyncGroup() { final String consumerId = "consumer"; subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Collections.<TopicPartition>emptyList(), Errors.REBALANCE_IN_PROGRESS.code())); client.prepareResponse(joinGroupFollowerResponse(2, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); assertFalse(subscriptions.partitionAssignmentNeeded()); assertEquals(Collections.singleton(tp), subscriptions.assignedPartitions()); }
@Test public void testIllegalGenerationOnSyncGroup() { final String consumerId = "consumer"; subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Collections.<TopicPartition>emptyList(), Errors.ILLEGAL_GENERATION.code())); client.prepareResponse(new MockClient.RequestMatcher() { @Override public boolean matches(ClientRequest request) { JoinGroupRequest joinRequest = new JoinGroupRequest(request.request().body()); return joinRequest.memberId().equals(JoinGroupRequest.UNKNOWN_MEMBER_ID); } }, joinGroupFollowerResponse(2, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); assertFalse(subscriptions.partitionAssignmentNeeded()); assertEquals(Collections.singleton(tp), subscriptions.assignedPartitions()); }
@Test public void testMetadataChangeTriggersRebalance() { final String consumerId = "consumer"; subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); assertFalse(subscriptions.partitionAssignmentNeeded()); metadata.update(TestUtils.singletonCluster(topicName, 2), time.milliseconds()); assertTrue(subscriptions.partitionAssignmentNeeded()); }
@Test public void testUpdateMetadataDuringRebalance() { final String topic1 = "topic1"; final String topic2 = "topic2"; TopicPartition tp1 = new TopicPartition(topic1, 0); TopicPartition tp2 = new TopicPartition(topic2, 0); final String consumerId = "leader"; List<String> topics = Arrays.asList(topic1, topic2); subscriptions.subscribe(topics, rebalanceListener); metadata.setTopics(topics); subscriptions.needReassignment(); metadata.update(TestUtils.singletonCluster(topic1, 1), time.milliseconds()); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, topics); partitionAssignor.prepare(Collections.singletonMap(consumerId, Arrays.asList(tp1))); 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()); if (sync.memberId().equals(consumerId) && sync.generationId() == 1 && sync.groupAssignment().containsKey(consumerId)) { Map<String, Integer> topicPartitionCounts = new HashMap<>(); topicPartitionCounts.put(topic1, 1); topicPartitionCounts.put(topic2, 1); metadata.update(TestUtils.singletonCluster(topicPartitionCounts), time.milliseconds()); return true; } return false; } }, syncGroupResponse(Arrays.asList(tp1), Errors.NONE.code())); client.prepareResponse(joinGroupLeaderResponse(2, consumerId, memberSubscriptions, Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp1, tp2), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); assertFalse(subscriptions.partitionAssignmentNeeded()); assertEquals(new HashSet<>(Arrays.asList(tp1, tp2)), subscriptions.assignedPartitions()); }
@Test public void testRejoinGroup() { subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); assertEquals(1, rebalanceListener.revokedCount); assertEquals(1, rebalanceListener.assignedCount); subscriptions.needReassignment(); client.prepareResponse(joinGroupFollowerResponse(2, "consumer", "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); assertEquals(2, rebalanceListener.revokedCount); assertEquals(Collections.singleton(tp), rebalanceListener.revoked); assertEquals(2, rebalanceListener.assignedCount); assertEquals(Collections.singleton(tp), rebalanceListener.assigned); }
@Test public void testDisconnectInJoin() { subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()), true); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code())); client.prepareResponse(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); }
@Test(expected = ApiException.class) public void testInvalidSessionTimeout() { subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(0, "consumer", "", Errors.INVALID_SESSION_TIMEOUT.code())); coordinator.ensurePartitionAssignment(); }
@Test public void testAutoCommitDynamicAssignment() { final String consumerId = "consumer"; ConsumerCoordinator coordinator = buildCoordinator(new Metrics(), assignors, ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, true); subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); subscriptions.seek(tp, 100); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); time.sleep(autoCommitIntervalMs); consumerClient.poll(0); assertEquals(100L, subscriptions.committed(tp).offset()); }
@Test public void testAutoCommitDynamicAssignmentRebalance() { final String consumerId = "consumer"; ConsumerCoordinator coordinator = buildCoordinator(new Metrics(), assignors, ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, true); subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.needReassignment(); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); time.sleep(autoCommitIntervalMs); consumerClient.poll(0); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code())); coordinator.ensurePartitionAssignment(); subscriptions.seek(tp, 100); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); time.sleep(autoCommitIntervalMs); consumerClient.poll(0); assertEquals(100L, subscriptions.committed(tp).offset()); }