@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()); }
public void ensurePartitionAssignment() { if (subscriptions.partitionsAutoAssigned()) { if (subscriptions.hasPatternSubscription()) { client.ensureFreshMetadata(); } ensureActiveGroup(); } }
ConsumerCoordinator extends AbstractCoordinator { public void ensurePartitionAssignment() { if (subscriptions.partitionsAutoAssigned()) { if (subscriptions.hasPatternSubscription()) { client.ensureFreshMetadata(); } ensureActiveGroup(); } } }
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); }
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); }
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);  }