@Test public void testProtocolMetadataOrder() { RoundRobinAssignor roundRobin = new RoundRobinAssignor(); RangeAssignor range = new RangeAssignor(); try (Metrics metrics = new Metrics(time)) { ConsumerCoordinator coordinator = buildCoordinator(metrics, Arrays.<PartitionAssignor>asList(roundRobin, range), ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, false); List<ProtocolMetadata> metadata = coordinator.metadata(); assertEquals(2, metadata.size()); assertEquals(roundRobin.name(), metadata.get(0).name()); assertEquals(range.name(), metadata.get(1).name()); } try (Metrics metrics = new Metrics(time)) { ConsumerCoordinator coordinator = buildCoordinator(metrics, Arrays.<PartitionAssignor>asList(range, roundRobin), ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, false); List<ProtocolMetadata> metadata = coordinator.metadata(); assertEquals(2, metadata.size()); assertEquals(range.name(), metadata.get(0).name()); assertEquals(roundRobin.name(), metadata.get(1).name()); } }
@Override public List<ProtocolMetadata> metadata() { List<ProtocolMetadata> metadataList = new ArrayList<>(); for (PartitionAssignor assignor : assignors) { Subscription subscription = assignor.subscription(subscriptions.subscription()); ByteBuffer metadata = ConsumerProtocol.serializeSubscription(subscription); metadataList.add(new ProtocolMetadata(assignor.name(), metadata)); } return metadataList; }
ConsumerCoordinator extends AbstractCoordinator { @Override public List<ProtocolMetadata> metadata() { List<ProtocolMetadata> metadataList = new ArrayList<>(); for (PartitionAssignor assignor : assignors) { Subscription subscription = assignor.subscription(subscriptions.subscription()); ByteBuffer metadata = ConsumerProtocol.serializeSubscription(subscription); metadataList.add(new ProtocolMetadata(assignor.name(), metadata)); } return metadataList; } }
ConsumerCoordinator extends AbstractCoordinator { @Override public List<ProtocolMetadata> metadata() { List<ProtocolMetadata> metadataList = new ArrayList<>(); for (PartitionAssignor assignor : assignors) { Subscription subscription = assignor.subscription(subscriptions.subscription()); ByteBuffer metadata = ConsumerProtocol.serializeSubscription(subscription); metadataList.add(new ProtocolMetadata(assignor.name(), metadata)); } return metadataList; } 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 { @Override public List<ProtocolMetadata> metadata() { List<ProtocolMetadata> metadataList = new ArrayList<>(); for (PartitionAssignor assignor : assignors) { Subscription subscription = assignor.subscription(subscriptions.subscription()); ByteBuffer metadata = ConsumerProtocol.serializeSubscription(subscription); metadataList.add(new ProtocolMetadata(assignor.name(), metadata)); } return metadataList; } 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 { @Override public List<ProtocolMetadata> metadata() { List<ProtocolMetadata> metadataList = new ArrayList<>(); for (PartitionAssignor assignor : assignors) { Subscription subscription = assignor.subscription(subscriptions.subscription()); ByteBuffer metadata = ConsumerProtocol.serializeSubscription(subscription); metadataList.add(new ProtocolMetadata(assignor.name(), metadata)); } return metadataList; } 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);  }