Github user liurenjie1024 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1835#discussion_r93559465
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/fetcher/ManualKafkaRecordsFetcher.java
---
@@ -0,0 +1,85 @@
+package org.apache.storm.kafka.spout.internal.fetcher;
+
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import
org.apache.storm.kafka.spout.internal.partition.KafkaPartitionReader;
+import org.apache.storm.kafka.spout.KafkaSpout;
+import org.apache.storm.kafka.spout.TopicPartitionComparator;
+
+import java.util.*;
+
+public class ManualKafkaRecordsFetcher<K, V> implements
KafkaRecordsFetcher<K, V> {
+ private static final Comparator<TopicPartition>
KAFKA_TOPIC_PARTITION_COMPARATOR = new TopicPartitionComparator();
+
+ private final KafkaConsumer<K, V> consumer;
+ private final int thisTaskIndex;
+ private final int totalTaskCount;
+ private final KafkaPartitionReader partitionReader;
+ private final KafkaSpout.Timer partitionRefreshTimer;
+ private final PartitionAssignmentChangeListener
partitionAssignmentChangeListener;
+ private Set<TopicPartition> myPartitions = Collections.emptySet();
+
+ public ManualKafkaRecordsFetcher(KafkaConsumer<K, V> consumer,
+ int thisTaskIndex,
+ int totalTaskCount,
+ KafkaPartitionReader partitionReader,
+ KafkaSpout.Timer
partitionRefreshTimer,
+ PartitionAssignmentChangeListener
partitionAssignmentChangeListener) {
+ this.consumer = consumer;
+ this.thisTaskIndex = thisTaskIndex;
+ this.totalTaskCount = totalTaskCount;
+ this.partitionReader = partitionReader;
+ this.partitionRefreshTimer = partitionRefreshTimer;
+ this.partitionAssignmentChangeListener =
partitionAssignmentChangeListener;
+
+ doRefreshMyPartitions();
+ }
+
+ private void refreshMyPartitionsIfNeeded() {
+ if (!partitionRefreshTimer.isExpiredResetOnTrue()) {
+ return;
+ }
+
+ doRefreshMyPartitions();
+ }
+
+ private void doRefreshMyPartitions() {
+ List<TopicPartition> topicPartitions =
partitionReader.readPartitions(consumer);
+ Collections.sort(topicPartitions,
KAFKA_TOPIC_PARTITION_COMPARATOR);
--- End diff --
This sort ensures that all instances of KafkaSpout see the same ordered
list so that the calculation is correct.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
