@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()); }
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);  }