[
https://issues.apache.org/jira/browse/FLINK-3102?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15045101#comment-15045101
]
ASF GitHub Bot commented on FLINK-3102:
---------------------------------------
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/1437#discussion_r46835390
--- Diff:
flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java
---
@@ -303,25 +300,29 @@ public FlinkKafkaConsumer(String topic,
KeyedDeserializationSchema<T> deserializ
validateZooKeeperConfig(props);
}
- // Connect to a broker to get the partitions
- List<PartitionInfo> partitionInfos =
getPartitionsForTopic(topic, props);
+ // Connect to a broker to get the partitions for all topics
+ this.partitionInfos = getPartitionsForTopic(topics, props);
if (partitionInfos.size() == 0) {
- throw new RuntimeException("Unable to retrieve any
partitions for topic " + topic + "." +
+ throw new RuntimeException("Unable to retrieve any
partitions for the requested topics " + StringUtils.join(topics, ",") + "." +
"Please check previous log entries");
}
- // get initial partitions list. The order of the partitions is
important for consistent
- // partition id assignment in restart cases.
- this.partitions = new int[partitionInfos.size()];
- for (int i = 0; i < partitionInfos.size(); i++) {
- partitions[i] = partitionInfos.get(i).partition();
-
- if (partitions[i] >= partitions.length) {
- throw new RuntimeException("Kafka partition
numbers are sparse");
+ Map<String, Integer> countPerTopic = new HashMap<>();
--- End diff --
How about guarding this section with `if (LOG.isInfoEnabled())`?
> Allow reading from multiple topics with one FlinkKafkaConsumer
> --------------------------------------------------------------
>
> Key: FLINK-3102
> URL: https://issues.apache.org/jira/browse/FLINK-3102
> Project: Flink
> Issue Type: New Feature
> Components: Kafka Connector
> Reporter: Robert Metzger
> Assignee: Robert Metzger
>
> Currently, a Kafka consumer allows to read from only one topic.
> For cases where multiple topics contain messages with the same schema, it is
> useful to allow to subscribe to many topics using one FlinkKafkaConsumer
> instance.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
