ConsumerCoordinator extends AbstractCoordinator { @Override public boolean needRejoin() { return subscriptions.partitionsAutoAssigned() && (super.needRejoin() || subscriptions.partitionAssignmentNeeded()); } 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 public void testIllegalGeneration() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.assignFromSubscribed(Collections.singletonList(tp)); time.sleep(sessionTimeoutMs); RequestFuture<Void> future = coordinator.sendHeartbeatRequest(); assertEquals(1, consumerClient.pendingRequestCount()); assertFalse(future.isDone()); client.prepareResponse(heartbeatResponse(Errors.ILLEGAL_GENERATION.code())); time.sleep(sessionTimeoutMs); consumerClient.poll(0); assertTrue(future.isDone()); assertTrue(future.failed()); assertEquals(Errors.ILLEGAL_GENERATION.exception(), future.exception()); assertTrue(coordinator.needRejoin()); }
@Test public void testUnknownConsumerId() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener); subscriptions.assignFromSubscribed(Collections.singletonList(tp)); time.sleep(sessionTimeoutMs); RequestFuture<Void> future = coordinator.sendHeartbeatRequest(); assertEquals(1, consumerClient.pendingRequestCount()); assertFalse(future.isDone()); client.prepareResponse(heartbeatResponse(Errors.UNKNOWN_MEMBER_ID.code())); time.sleep(sessionTimeoutMs); consumerClient.poll(0); assertTrue(future.isDone()); assertTrue(future.failed()); assertEquals(Errors.UNKNOWN_MEMBER_ID.exception(), future.exception()); assertTrue(coordinator.needRejoin()); }