[
https://issues.apache.org/jira/browse/FLINK-3102?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15045135#comment-15045135
]
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_r46837841
--- 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) {
+ long offset =
currentOffsets.get(tp);
+ Long lastCommitted =
(Long)consumer.commitedOffsets.get(tp);
+ if(lastCommitted ==
null) {
+ lastCommitted =
OFFSET_NOT_SET;
+ }
if (offset !=
OFFSET_NOT_SET) {
if (offset >
lastCommitted) {
offsetsToCommit.put(tp, offset);
-
LOG.debug("Committing offset {} for partition {}", offset, partition);
+
//noinspection unchecked
+
consumer.commitedOffsets.put(tp, offset);
+
LOG.debug("Committing offset {} for partition {}", offset, tp);
} else {
-
LOG.debug("Ignoring offset {} for partition {} because it is already
committed", offset, partition);
+
LOG.debug("Ignoring offset {} for partition {} because it is already
committed", offset, tp);
}
}
}
+ if(LOG.isInfoEnabled() &&
offsetsToCommit.size() > 0) {
--- End diff --
Same as above: Info logging may be pretty aggressive...
> 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)
