Repository: storm Updated Branches: refs/heads/master cef450064 -> 5077228df
STORM-2784: storm-kafka-client KafkaTupleListener method onPartitionsReassigned() should be called after initialization is complete Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/ae827dac Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/ae827dac Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/ae827dac Branch: refs/heads/master Commit: ae827dac08fa67997c5eb8e6c23828f190de2851 Parents: 7b940ae Author: Hugo Louro <[email protected]> Authored: Sun Oct 22 21:35:29 2017 -0700 Committer: Hugo Louro <[email protected]> Committed: Sun Oct 22 21:49:00 2017 -0700 ---------------------------------------------------------------------- .../src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/ae827dac/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java index 9253a2d..4fb5e31 100644 --- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java +++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java @@ -165,8 +165,8 @@ public class KafkaSpout<K, V> extends BaseRichSpout { LOG.info("Partitions reassignment. [task-ID={}, consumer-group={}, consumer={}, topic-partitions={}]", context.getThisTaskId(), kafkaSpoutConfig.getConsumerGroupId(), kafkaConsumer, partitions); - tupleListener.onPartitionsReassigned(partitions); initialize(partitions); + tupleListener.onPartitionsReassigned(partitions); } private void initialize(Collection<TopicPartition> partitions) {
