[
https://issues.apache.org/jira/browse/FLINK-3102?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15045156#comment-15045156
]
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_r46839322
--- Diff:
flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java
---
@@ -640,30 +642,38 @@ public PeriodicOffsetCommitter(long commitInterval,
FlinkKafkaConsumer consumer)
@Override
public void run() {
try {
+
while (running) {
try {
Thread.sleep(commitInterval);
// ------------ commit
current offsets ----------------
// create copy of current
offsets
- long[] currentOffsets =
Arrays.copyOf(consumer.lastOffsets, consumer.lastOffsets.length);
-
- Map<TopicPartition, Long>
offsetsToCommit = new HashMap<>();
- for (TopicPartition tp :
(List<TopicPartition>)consumer.subscribedPartitions) {
- int partition =
tp.partition();
- long offset =
currentOffsets[partition];
- long lastCommitted =
consumer.commitedOffsets[partition];
-
+ HashMap<KafkaTopicPartition,
Long> currentOffsets = (HashMap<KafkaTopicPartition, Long>)
consumer.lastOffsets.clone();
+
+ Map<KafkaTopicPartition, Long>
offsetsToCommit = new HashMap<>();
+ for (KafkaTopicPartition tp :
(List<KafkaTopicPartition>)consumer.subscribedPartitions) {
--- End diff --
A bit unclean to cast here from `List<KafkaTopicPartitionLeader>` to
`List<KafkaTopicPartition>`. Why not simply iterate with the
`KafkaTopicPartitionLeader` type?
> 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)
