Skip updating statistics for Kafka partitions in KafkaExtractor when topic is skipped.[]
Closes from sv2000/kafkaPartitionIndex
@@ -370,8 +370,9 @@ public abstract class KafkaExtractor<S, D> extends EventBasedExtractor<S, D> {
@Override
public void close() throws IOException {
-
+ if (currentPartitionIdx != INITIAL_PARTITION_IDX) {
updateStatisticsForCurrentPartition();
+ }
Map<KafkaPartition, Map<String, String>> tagsForPartitionsMap = Maps.newHashMap();
