[STORM-8019] Fixing kafka topic level metrics computation

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java

@@ -39,6 +39,7 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+
 import org.apache.commons.lang.Validate;
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -154,7 +155,7 @@ public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputC
 
         tupleListener.open(conf, context);
         this.kafkaOffsetMetricManager
-            = new KafkaOffsetMetricManager<>(() -> Collections.unmodifiableMap(offsetManagers), () -> admin, context);
+                = new KafkaOffsetMetricManager<>(() -> Collections.unmodifiableMap(offsetManagers), () -> admin, context);
 
         LOG.info("Kafka Spout opened with the following configuration: {}", kafkaSpoutConfig);
     }
@@ -183,7 +184,7 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
             previousAssignment = partitions;
 
             LOG.info("Partitions revoked. [consumer-group={}, consumer={}, topic-partitions={}]",
-                kafkaSpoutConfig.getConsumerGroupId(), consumer, partitions);
+                    kafkaSpoutConfig.getConsumerGroupId(), consumer, partitions);
 
             if (isAtLeastOnceProcessing()) {
                 commitOffsetsForAckedTuples();
@@ -193,7 +194,7 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
         @Override
         public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
             LOG.info("Partitions reassignment. [task-ID={}, consumer-group={}, consumer={}, topic-partitions={}]",
-                context.getThisTaskId(), kafkaSpoutConfig.getConsumerGroupId(), consumer, partitions);
+                    context.getThisTaskId(), kafkaSpoutConfig.getConsumerGroupId(), consumer, partitions);
 
             initialize(partitions);
             tupleListener.onPartitionsReassigned(partitions);
@@ -221,7 +222,7 @@ private void initialize(Collection<TopicPartition> partitions) {
                 final OffsetAndMetadata committedOffset = consumer.committed(newTp);
                 final long fetchOffset = doSeek(newTp, committedOffset);
                 LOG.debug("Set consumer position to [{}] for topic-partition [{}] with [{}] and committed offset [{}]",
-                    fetchOffset, newTp, firstPollOffsetStrategy, committedOffset);
+                        fetchOffset, newTp, firstPollOffsetStrategy, committedOffset);
                 if (isAtLeastOnceProcessing() && !offsetManagers.containsKey(newTp)) {
                     offsetManagers.put(newTp, new OffsetManager(newTp, fetchOffset));
                 }
@@ -234,13 +235,13 @@ private void initialize(Collection<TopicPartition> partitions) {
          */
         private long doSeek(TopicPartition newTp, OffsetAndMetadata committedOffset) {
             LOG.trace("Seeking offset for topic-partition [{}] with [{}] and committed offset [{}]",
-                newTp, firstPollOffsetStrategy, committedOffset);
+                    newTp, firstPollOffsetStrategy, committedOffset);
 
             if (committedOffset != null) {
                 // offset was previously committed for this consumer group and topic-partition, either by this or another topology.
                 if (commitMetadataManager.isOffsetCommittedByThisTopology(newTp,
-                    committedOffset,
-                    Collections.unmodifiableMap(offsetManagers))) {
+                        committedOffset,
+                        Collections.unmodifiableMap(offsetManagers))) {
                     // Another KafkaSpout instance (of this topology) already committed, therefore FirstPollOffsetStrategy does not apply.
                     consumer.seek(newTp, committedOffset.offset());
                 } else {
@@ -281,7 +282,7 @@ public void nextTuple() {
                     commitOffsetsForAckedTuples();
                 } else if (kafkaSpoutConfig.getProcessingGuarantee() == ProcessingGuarantee.NO_GUARANTEE) {
                     Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
-                        createFetchedOffsetsMetadata(consumer.assignment());
+                            createFetchedOffsetsMetadata(consumer.assignment());
                     consumer.commitAsync(offsetsToCommit, null);
                     LOG.debug("Committed offsets {} to Kafka", offsetsToCommit);
                 }
@@ -336,7 +337,7 @@ private PollablePartitionsInfo getPollablePartitionsInfo() {
                     pollablePartitions.add(tp);
                 } else {
                     LOG.debug("Not polling on partition [{}]. It has [{}] uncommitted offsets, which exceeds the limit of [{}]. ", tp,
-                        numUncommittedOffsets, maxUncommittedOffsets);
+                            numUncommittedOffsets, maxUncommittedOffsets);
                 }
             }
         }
@@ -345,7 +346,7 @@ private PollablePartitionsInfo getPollablePartitionsInfo() {
 
     private boolean isWaitingToEmit() {
         return waitingToEmit.values().stream()
-            .anyMatch(list -> !list.isEmpty());
+                .anyMatch(list -> !list.isEmpty());
     }
 
     private void setWaitingToEmit(ConsumerRecords<K, V> consumerRecords) {
@@ -365,11 +366,11 @@ private ConsumerRecords<K, V> pollKafkaBroker(PollablePartitionsInfo pollablePar
             ackRetriableOffsetsIfCompactedAway(pollablePartitionsInfo.pollableEarliestRetriableOffsets, consumerRecords);
             final int numPolledRecords = consumerRecords.count();
             LOG.debug("Polled [{}] records from Kafka",
-                numPolledRecords);
+                    numPolledRecords);
             if (kafkaSpoutConfig.getProcessingGuarantee() == KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE) {
                 //Commit polled records immediately to ensure delivery is at-most-once.
                 Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
-                    createFetchedOffsetsMetadata(consumer.assignment());
+                        createFetchedOffsetsMetadata(consumer.assignment());
                 consumer.commitSync(offsetsToCommit);
                 LOG.debug("Committed offsets {} to Kafka", offsetsToCommit);
             }
@@ -387,7 +388,7 @@ private void doSeekRetriableTopicPartitions(Map<TopicPartition, Long> pollableEa
     }
 
     private void ackRetriableOffsetsIfCompactedAway(Map<TopicPartition, Long> earliestRetriableOffsets,
-        ConsumerRecords<K, V> consumerRecords) {
+                                                    ConsumerRecords<K, V> consumerRecords) {
         for (Entry<TopicPartition, Long> entry : earliestRetriableOffsets.entrySet()) {
             TopicPartition tp = entry.getKey();
             List<ConsumerRecord<K, V>> records = consumerRecords.records(tp);
@@ -529,7 +530,7 @@ private void commitOffsetsForAckedTuples() {
                      * to the committed offset.
                      */
                     LOG.debug("Consumer fell behind committed offset. Catching up. Position was [{}], skipping to [{}]",
-                        position, committedOffset);
+                            position, committedOffset);
                     consumer.seek(tp, committedOffset);
                 }
                 /**
@@ -539,8 +540,8 @@ private void commitOffsetsForAckedTuples() {
                 if (waitingToEmitForTp != null) {
                     //Discard the pending records that are already committed
                     waitingToEmit.put(tp, waitingToEmitForTp.stream()
-                        .filter(record -> record.offset() >= committedOffset)
-                        .collect(Collectors.toCollection(LinkedList::new)));
+                            .filter(record -> record.offset() >= committedOffset)
+                            .collect(Collectors.toCollection(LinkedList::new)));
                 }
 
                 final OffsetManager offsetManager = offsetManagers.get(tp);
@@ -573,11 +574,11 @@ public void ack(Object messageId) {
 
         if (!emitted.contains(msgId)) {
             LOG.debug("Received ack for message [{}], associated with tuple emitted for a ConsumerRecord that "
-                + "came from a topic-partition that this consumer group instance is no longer tracking "
-                + "due to rebalance/partition reassignment. No action taken.", msgId);
+                    + "came from a topic-partition that this consumer group instance is no longer tracking "
+                    + "due to rebalance/partition reassignment. No action taken.", msgId);
         } else {
             Validate.isTrue(!retryService.isScheduled(msgId), "The message id " + msgId + " is queued for retry while being acked."
-                + " This should never occur barring errors in the RetryService implementation or the spout code.");
+                    + " This should never occur barring errors in the RetryService implementation or the spout code.");
             offsetManagers.get(msgId.getTopicPartition()).addToAckMsgs(msgId);
             emitted.remove(msgId);
         }
@@ -595,11 +596,11 @@ public void fail(Object messageId) {
         final KafkaSpoutMessageId msgId = (KafkaSpoutMessageId) messageId;
         if (!emitted.contains(msgId)) {
             LOG.debug("Received fail for tuple this spout is no longer tracking."
-                + " Partitions may have been reassigned. Ignoring message [{}]", msgId);
+                    + " Partitions may have been reassigned. Ignoring message [{}]", msgId);
             return;
         }
         Validate.isTrue(!retryService.isScheduled(msgId), "The message id " + msgId + " is queued for retry while being failed."
-            + " This should never occur barring errors in the RetryService implementation or the spout code.");
+                + " This should never occur barring errors in the RetryService implementation or the spout code.");
 
         msgId.incrementNumFails();
 
@@ -630,7 +631,7 @@ private void refreshAssignment() {
         List<TopicPartition> allPartitionsSorted = new ArrayList<>(allPartitions);
         Collections.sort(allPartitionsSorted, TopicPartitionComparator.INSTANCE);
         Set<TopicPartition> assignedPartitions = kafkaSpoutConfig.getTopicPartitioner()
-            .getPartitionsForThisTask(allPartitionsSorted, context);
+                .getPartitionsForThisTask(allPartitionsSorted, context);
         boolean partitionChanged = topicAssigner.assignPartitions(consumer, assignedPartitions, rebalanceListener);
         if (partitionChanged && canRegisterMetrics()) {
             LOG.info("Partitions assignments has changed, updating metrics.");
@@ -683,9 +684,9 @@ public void declareOutputFields(OutputFieldsDeclarer declarer) {
     @Override
     public String toString() {
         return "KafkaSpout{"
-            + "offsetManagers =" + offsetManagers
-            + ", emitted=" + emitted
-            + "}";
+                + "offsetManagers =" + offsetManagers
+                + ", emitted=" + emitted
+                + "}";
     }
 
     @Override
@@ -718,8 +719,8 @@ private boolean isPrimitiveOrWrapper(Class<?> type) {
 
     private boolean isWrapper(Class<?> type) {
         return type == Double.class || type == Float.class || type == Long.class
-            || type == Integer.class || type == Short.class || type == Character.class
-            || type == Byte.class || type == Boolean.class || type == String.class;
+                || type == Integer.class || type == Short.class || type == Character.class
+                || type == Byte.class || type == Boolean.class || type == String.class;
     }
 
     private String getTopicsString() {
@@ -735,8 +736,8 @@ private static class PollablePartitionsInfo {
         PollablePartitionsInfo(Set<TopicPartition> pollablePartitions, Map<TopicPartition, Long> earliestRetriableOffsets) {
             this.pollablePartitions = pollablePartitions;
             this.pollableEarliestRetriableOffsets = earliestRetriableOffsets.entrySet().stream()
-                .filter(entry -> pollablePartitions.contains(entry.getKey()))
-                .collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue()));
+                    .filter(entry -> pollablePartitions.contains(entry.getKey()))
+                    .collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue()));
         }
 
         public boolean shouldPoll() {

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetMetricManager.java

@@ -6,9 +6,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- *
+ * <p>
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -22,14 +22,15 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Supplier;
-
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.storm.kafka.spout.internal.OffsetManager;
 import org.apache.storm.task.TopologyContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
+
 /**
  * This class is used to manage both the partition and topic level offset metrics.
  */
@@ -55,20 +56,22 @@ public KafkaOffsetMetricManager(Supplier<Map<TopicPartition, OffsetManager>> off
     }
 
     public void registerMetricsForNewTopicPartitions(Set<TopicPartition> newAssignment) {
+
         for (TopicPartition topicPartition : newAssignment) {
             if (!topicPartitionMetricsMap.containsKey(topicPartition)) {
                 LOG.info("Registering metric for topicPartition: {}", topicPartition);
                 // create topic level metrics for given topic if absent
                 String topic = topicPartition.topic();
                 KafkaOffsetTopicMetrics topicMetrics = topicMetricsMap.get(topic);
                 if (topicMetrics == null) {
-                    topicMetrics = new KafkaOffsetTopicMetrics(topic);
+                    LOG.info("Registering metric for topic: {}", topic);
+                    topicMetrics = new KafkaOffsetTopicMetrics(topic, offsetManagerSupplier, adminSupplier, newAssignment);
                     topicMetricsMap.put(topic, topicMetrics);
                     topologyContext.registerMetricSet("kafkaOffset", topicMetrics);
                 }
 
                 KafkaOffsetPartitionMetrics topicPartitionMetricSet
-                    = new KafkaOffsetPartitionMetrics<>(offsetManagerSupplier, adminSupplier, topicPartition, topicMetrics);
+                        = new KafkaOffsetPartitionMetrics<>(offsetManagerSupplier, adminSupplier, topicPartition);
                 topicPartitionMetricsMap.put(topicPartition, topicPartitionMetricSet);
                 topologyContext.registerMetricSet("kafkaOffset", topicPartitionMetricSet);
             }