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 a328aec5d70..0fcb22a0446 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
@@ -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() {
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetMetricManager.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetMetricManager.java
index 2b150797ffe..64c42832cf3 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetMetricManager.java
+++ b/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,7 +22,6 @@
 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;
@@ -30,6 +29,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
+
 /**
  * This class is used to manage both the partition and topic level offset metrics.
  */
@@ -55,6 +56,7 @@ 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);
@@ -62,13 +64,14 @@ public void registerMetricsForNewTopicPartitions(Set<TopicPartition> newAssignme
                 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);
             }
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetPartitionMetrics.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetPartitionMetrics.java
index 6b29eefc919..df5b0c8694e 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetPartitionMetrics.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetPartitionMetrics.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.
@@ -18,27 +18,25 @@
 
 package org.apache.storm.kafka.spout.metrics2;
 
+import static org.apache.storm.kafka.spout.metrics2.KafkaOffsetUtil.getBeginningOffsets;
+import static org.apache.storm.kafka.spout.metrics2.KafkaOffsetUtil.getEndOffsets;
+
 import com.codahale.metrics.Gauge;
 import com.codahale.metrics.Metric;
 import com.codahale.metrics.MetricSet;
-
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.function.Supplier;
-
 import org.apache.kafka.clients.admin.Admin;
-import org.apache.kafka.clients.admin.ListOffsetsResult;
-import org.apache.kafka.clients.admin.OffsetSpec;
-import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.RetriableException;
 import org.apache.storm.kafka.spout.internal.OffsetManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
+
+
 /**
  * Partition level metrics.
  * <p>
@@ -56,16 +54,13 @@ public class KafkaOffsetPartitionMetrics<K, V> implements MetricSet {
     private final Supplier<Admin> adminSupplier;
 
     private TopicPartition topicPartition;
-    private KafkaOffsetTopicMetrics topicMetrics;
 
     public KafkaOffsetPartitionMetrics(Supplier<Map<TopicPartition, OffsetManager>> offsetManagerSupplier,
                                        Supplier<Admin> adminSupplier,
-                                       TopicPartition topicPartition,
-                                       KafkaOffsetTopicMetrics topicMetrics) {
+                                       TopicPartition topicPartition) {
         this.offsetManagerSupplier = offsetManagerSupplier;
         this.adminSupplier = adminSupplier;
         this.topicPartition = topicPartition;
-        this.topicMetrics = topicMetrics;
 
         LOG.info("Running KafkaOffsetMetricSet");
     }
@@ -74,93 +69,57 @@ public KafkaOffsetPartitionMetrics(Supplier<Map<TopicPartition, OffsetManager>>
     public Map<String, Metric> getMetrics() {
         Map<String, Metric> metrics = new HashMap();
 
-        String metricPath = topicPartition.topic()  + "/partition_" + topicPartition.partition();
-        Gauge<Long> spoutLagGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                Map<TopicPartition, Long> endOffsets = getEndOffsets(Collections.singleton(topicPartition));
-                if (endOffsets == null || endOffsets.isEmpty()) {
-                    LOG.error("Failed to get endOffsets from Kafka for topic partitions: {}.", topicPartition);
-                    return 0L;
-                }
-                // add value to topic level metric
-                OffsetManager offsetManager = offsetManagerSupplier.get().get(topicPartition);
-                Long ret = endOffsets.get(topicPartition) - offsetManager.getCommittedOffset();
-                topicMetrics.totalSpoutLag += ret;
-                return ret;
+        String metricPath = topicPartition.topic() + "/partition_" + topicPartition.partition();
+        Gauge<Long> spoutLagGauge = () -> {
+            Map<TopicPartition, Long> endOffsets = getEndOffsets(Collections.singleton(topicPartition), adminSupplier);
+            if (endOffsets == null || endOffsets.isEmpty()) {
+                LOG.error("Failed to get endOffsets from Kafka for topic partitions: {}.", topicPartition);
+                return 0L;
             }
+            OffsetManager offsetManager = offsetManagerSupplier.get().get(topicPartition);
+            return endOffsets.get(topicPartition) - offsetManager.getCommittedOffset();
         };
 
-        Gauge<Long> earliestTimeOffsetGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                Map<TopicPartition, Long> beginningOffsets = getBeginningOffsets(Collections.singleton(topicPartition));
-                if (beginningOffsets == null || beginningOffsets.isEmpty()) {
-                    LOG.error("Failed to get beginningOffsets from Kafka for topic partitions: {}.", topicPartition);
-                    return 0L;
-                }
-                // add value to topic level metric
-                Long ret = beginningOffsets.get(topicPartition);
-                topicMetrics.totalEarliestTimeOffset += beginningOffsets.get(topicPartition);
-                return ret;
+        Gauge<Long> earliestTimeOffsetGauge = () -> {
+            Map<TopicPartition, Long> beginningOffsets = getBeginningOffsets(Collections.singleton(topicPartition), adminSupplier);
+            if (beginningOffsets == null || beginningOffsets.isEmpty()) {
+                LOG.error("Failed to get beginningOffsets from Kafka for topic partitions: {}.", topicPartition);
+                return 0L;
             }
+            return beginningOffsets.get(topicPartition);
         };
 
-        Gauge<Long> latestTimeOffsetGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                Map<TopicPartition, Long> endOffsets = getEndOffsets(Collections.singleton(topicPartition));
-                if (endOffsets == null || endOffsets.isEmpty()) {
-                    LOG.error("Failed to get endOffsets from Kafka for topic partitions: {}.", topicPartition);
-                    return 0L;
-                }
-                // add value to topic level metric
-                Long ret = endOffsets.get(topicPartition);
-                topicMetrics.totalLatestTimeOffset += ret;
-                return ret;
+        Gauge<Long> latestTimeOffsetGauge = () -> {
+            Map<TopicPartition, Long> endOffsets = getEndOffsets(Collections.singleton(topicPartition), adminSupplier);
+            if (endOffsets == null || endOffsets.isEmpty()) {
+                LOG.error("Failed to get endOffsets from Kafka for topic partitions: {}.", topicPartition);
+                return 0L;
             }
+            return endOffsets.get(topicPartition);
         };
 
-        Gauge<Long> latestEmittedOffsetGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                // add value to topic level metric
-                OffsetManager offsetManager = offsetManagerSupplier.get().get(topicPartition);
-                Long ret = offsetManager.getLatestEmittedOffset();
-                topicMetrics.totalLatestEmittedOffset += ret;
-                return ret;
-            }
+        Gauge<Long> latestEmittedOffsetGauge = () -> {
+            OffsetManager offsetManager = offsetManagerSupplier.get().get(topicPartition);
+            return offsetManager.getLatestEmittedOffset();
         };
 
-        Gauge<Long> latestCompletedOffsetGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                // add value to topic level metric
-                OffsetManager offsetManager = offsetManagerSupplier.get().get(topicPartition);
-                Long ret = offsetManager.getCommittedOffset();
-                topicMetrics.totalLatestCompletedOffset += ret;
-                return ret;
-            }
+        Gauge<Long> latestCompletedOffsetGauge = () -> {
+            OffsetManager offsetManager = offsetManagerSupplier.get().get(topicPartition);
+            return offsetManager.getCommittedOffset();
         };
 
-        Gauge<Long> recordsInPartitionGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                Map<TopicPartition, Long> endOffsets = getEndOffsets(Collections.singleton(topicPartition));
-                if (endOffsets == null || endOffsets.isEmpty()) {
-                    LOG.error("Failed to get endOffsets from Kafka for topic partitions: {}.", topicPartition);
-                    return 0L;
-                }
-                Map<TopicPartition, Long> beginningOffsets = getBeginningOffsets(Collections.singleton(topicPartition));
-                if (beginningOffsets == null || beginningOffsets.isEmpty()) {
-                    LOG.error("Failed to get beginningOffsets from Kafka for topic partitions: {}.", topicPartition);
-                    return 0L;
-                }
-                // add value to topic level metric
-                Long ret = endOffsets.get(topicPartition) - beginningOffsets.get(topicPartition);
-                topicMetrics.totalRecordsInPartitions += ret;
-                return ret;
+        Gauge<Long> recordsInPartitionGauge = () -> {
+            Map<TopicPartition, Long> endOffsets = getEndOffsets(Collections.singleton(topicPartition), adminSupplier);
+            if (endOffsets == null || endOffsets.isEmpty()) {
+                LOG.error("Failed to get endOffsets from Kafka for topic partitions: {}.", topicPartition);
+                return 0L;
+            }
+            Map<TopicPartition, Long> beginningOffsets = getBeginningOffsets(Collections.singleton(topicPartition), adminSupplier);
+            if (beginningOffsets == null || beginningOffsets.isEmpty()) {
+                LOG.error("Failed to get beginningOffsets from Kafka for topic partitions: {}.", topicPartition);
+                return 0L;
             }
+            return endOffsets.get(topicPartition) - beginningOffsets.get(topicPartition);
         };
 
         metrics.put(metricPath + "/" + "spoutLag", spoutLagGauge);
@@ -172,56 +131,4 @@ public Long getValue() {
 
         return metrics;
     }
-
-    private Map<TopicPartition, Long> getBeginningOffsets(Set<TopicPartition> topicPartitions) {
-        Admin admin = adminSupplier.get();
-        if (admin == null) {
-            LOG.error("Kafka admin object is null, returning 0.");
-            return Collections.EMPTY_MAP;
-        }
-
-        Map<TopicPartition, Long> beginningOffsets;
-        try {
-            beginningOffsets = getOffsets(admin, topicPartitions, OffsetSpec.earliest());
-        } catch (RetriableException | ExecutionException | InterruptedException e) {
-            LOG.error("Failed to get offset from Kafka for topic partitions: {}.", topicPartition, e);
-            return Collections.EMPTY_MAP;
-        }
-        return beginningOffsets;
-    }
-
-    private Map<TopicPartition, Long> getEndOffsets(Set<TopicPartition> topicPartitions) {
-        Admin admin = adminSupplier.get();
-        if (admin == null) {
-            LOG.error("Kafka admin object is null, returning 0.");
-            return Collections.EMPTY_MAP;
-        }
-
-        Map<TopicPartition, Long> endOffsets;
-        try {
-            endOffsets = getOffsets(admin, topicPartitions, OffsetSpec.latest());
-        } catch (RetriableException | ExecutionException | InterruptedException e) {
-            LOG.error("Failed to get offset from Kafka for topic partitions: {}.", topicPartition, e);
-            return Collections.EMPTY_MAP;
-        }
-        return endOffsets;
-    }
-
-    private static Map<TopicPartition, Long> getOffsets(Admin admin, Set<TopicPartition> topicPartitions, OffsetSpec offsetSpec)
-        throws InterruptedException, ExecutionException {
-
-        Map<TopicPartition, OffsetSpec> offsetSpecMap = new HashMap<>();
-        for (TopicPartition topicPartition : topicPartitions) {
-            offsetSpecMap.put(topicPartition, offsetSpec);
-        }
-        Map<TopicPartition, Long> ret = new HashMap<>();
-        ListOffsetsResult listOffsetsResult = admin.listOffsets(offsetSpecMap);
-        KafkaFuture<Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo>> all = listOffsetsResult.all();
-        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionListOffsetsResultInfoMap = all.get();
-        for (Map.Entry<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> entry :
-                topicPartitionListOffsetsResultInfoMap.entrySet()) {
-            ret.put(entry.getKey(), entry.getValue().offset());
-        }
-        return ret;
-    }
 }
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetTopicMetrics.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetTopicMetrics.java
index cd0fbfc6b2c..8bdcef37485 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetTopicMetrics.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetTopicMetrics.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.
@@ -18,14 +18,26 @@
 
 package org.apache.storm.kafka.spout.metrics2;
 
+import static org.apache.storm.kafka.spout.metrics2.KafkaOffsetUtil.getBeginningOffsets;
+import static org.apache.storm.kafka.spout.metrics2.KafkaOffsetUtil.getEndOffsets;
+
 import com.codahale.metrics.Gauge;
 import com.codahale.metrics.Metric;
 import com.codahale.metrics.MetricSet;
+
+import java.util.Collections;
 import java.util.HashMap;
 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.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
 /**
  * Topic level metrics.
  * <p>
@@ -42,22 +54,18 @@ public class KafkaOffsetTopicMetrics implements MetricSet {
     private static final Logger LOG = LoggerFactory.getLogger(KafkaOffsetTopicMetrics.class);
 
     private String topic;
-    long totalSpoutLag;
-    long totalEarliestTimeOffset;
-    long totalLatestTimeOffset;
-    long totalLatestEmittedOffset;
-    long totalLatestCompletedOffset;
-    long totalRecordsInPartitions;
-
-
-    public KafkaOffsetTopicMetrics(String topic) {
+    Set<TopicPartition> assignment;
+    Supplier<Map<TopicPartition, OffsetManager>> offsetManagerSupplier;
+    Supplier<Admin> adminSupplier;
+
+    public KafkaOffsetTopicMetrics(String topic, Supplier<Map<TopicPartition,
+                                           OffsetManager>> offsetManagerSupplier,
+                                   Supplier<Admin> adminSupplier,
+                                   Set<TopicPartition> newAssignment) {
         this.topic = topic;
-        this.totalSpoutLag = 0L;
-        this.totalEarliestTimeOffset = 0L;
-        this.totalLatestTimeOffset = 0L;
-        this.totalLatestEmittedOffset = 0L;
-        this.totalLatestCompletedOffset = 0L;
-        this.totalRecordsInPartitions = 0L;
+        this.assignment = newAssignment;
+        this.offsetManagerSupplier = offsetManagerSupplier;
+        this.adminSupplier = adminSupplier;
         LOG.info("Create KafkaOffsetTopicMetrics for topic: {}", topic);
     }
 
@@ -65,46 +73,117 @@ public KafkaOffsetTopicMetrics(String topic) {
     public Map<String, Metric> getMetrics() {
         Map<String, Metric> metrics = new HashMap();
 
-        Gauge<Long> totalSpoutLagGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                return totalSpoutLag;
+        Gauge<Long> totalSpoutLagGauge = () -> {
+            Long totalSpoutLag = 0L;
+            for (TopicPartition topicPartition : assignment) {
+                String topicOfPartition = topicPartition.topic();
+                if (topicOfPartition.equals(topic)) {
+                    Map<TopicPartition, Long> endOffsets = getEndOffsets(Collections.singleton(topicPartition), adminSupplier);
+                    if (endOffsets == null || endOffsets.isEmpty()) {
+                        LOG.error("Failed to get endOffsets from Kafka for topic partitions: {}.", topicPartition);
+                        return 0L;
+                    }
+                    // add value to topic level metric
+                    OffsetManager offsetManager = offsetManagerSupplier.get().get(topicPartition);
+                    Long ret = endOffsets.get(topicPartition) - offsetManager.getCommittedOffset();
+                    totalSpoutLag += ret;
+                }
             }
+            return totalSpoutLag;
         };
 
-        Gauge<Long> totalEarliestTimeOffsetGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                return totalEarliestTimeOffset;
+        Gauge<Long> totalEarliestTimeOffsetGauge = () -> {
+
+            Long totalEarliestTimeOffset = 0L;
+
+            for (TopicPartition topicPartition : assignment) {
+                String topicOfPartition = topicPartition.topic();
+                if (topicOfPartition.equals(topic)) {
+                    Map<TopicPartition, Long> beginningOffsets = getBeginningOffsets(Collections.singleton(topicPartition), adminSupplier);
+                    if (beginningOffsets == null || beginningOffsets.isEmpty()) {
+                        LOG.error("Failed to get beginningOffsets from Kafka for topic partitions: {}.", topicPartition);
+                        return 0L;
+                    }
+                    // add value to topic level metric
+                    Long ret = beginningOffsets.get(topicPartition);
+                    totalEarliestTimeOffset += ret;
+                }
+
             }
+            return totalEarliestTimeOffset;
         };
 
-        Gauge<Long> totalLatestTimeOffsetGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                return totalLatestTimeOffset;
+        Gauge<Long> totalLatestTimeOffsetGauge = () -> {
+
+            Long totalLatestTimeOffset = 0L;
+            for (TopicPartition topicPartition : assignment) {
+                String topicOfPartition = topicPartition.topic();
+                if (topicOfPartition.equals(topic)) {
+                    Map<TopicPartition, Long> endOffsets = getEndOffsets(Collections.singleton(topicPartition), adminSupplier);
+                    if (endOffsets == null || endOffsets.isEmpty()) {
+                        LOG.error("Failed to get endOffsets from Kafka for topic partitions: {}.", topicPartition);
+                        return 0L;
+                    }
+                    // add value to topic level metric
+                    Long ret = endOffsets.get(topicPartition);
+                    totalLatestTimeOffset += ret;
+                }
             }
+            return totalLatestTimeOffset;
         };
 
-        Gauge<Long> totalLatestEmittedOffsetGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                return totalLatestEmittedOffset;
+        Gauge<Long> totalLatestEmittedOffsetGauge = () -> {
+
+            Long totalLatestEmittedOffset = 0L;
+            for (TopicPartition topicPartition : assignment) {
+                String topicOfPartition = topicPartition.topic();
+                if (topicOfPartition.equals(topic)) {
+                    OffsetManager offsetManager = offsetManagerSupplier.get().get(topicPartition);
+                    Long ret = offsetManager.getLatestEmittedOffset();
+                    totalLatestEmittedOffset += ret;
+                }
+
             }
+            return totalLatestEmittedOffset;
         };
 
-        Gauge<Long> totalLatestCompletedOffsetGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                return totalLatestCompletedOffset;
+        Gauge<Long> totalLatestCompletedOffsetGauge = () -> {
+
+            Long totalLatestCompletedOffset = 0L;
+            for (TopicPartition topicPartition : assignment) {
+                String topicOfPartition = topicPartition.topic();
+                if (topicOfPartition.equals(topic)) {
+                    // add value to topic level metric
+                    OffsetManager offsetManager = offsetManagerSupplier.get().get(topicPartition);
+                    Long ret = offsetManager.getCommittedOffset();
+                    totalLatestCompletedOffset += ret;
+                }
             }
+
+            return totalLatestCompletedOffset;
         };
 
-        Gauge<Long> totalRecordsInPartitionsGauge = new Gauge<Long>() {
-            @Override
-            public Long getValue() {
-                return totalRecordsInPartitions;
+        Gauge<Long> totalRecordsInPartitionsGauge = () -> {
+            Long totalRecordsInPartitions = 0L;
+            for (TopicPartition topicPartition : assignment) {
+                String topicOfPartition = topicPartition.topic();
+                if (topicOfPartition.equals(topic)) {
+                    Map<TopicPartition, Long> endOffsets = getEndOffsets(Collections.singleton(topicPartition), adminSupplier);
+                    if (endOffsets == null || endOffsets.isEmpty()) {
+                        LOG.error("Failed to get endOffsets from Kafka for topic partitions: {}.", topicPartition);
+                        return 0L;
+                    }
+                    Map<TopicPartition, Long> beginningOffsets = getBeginningOffsets(Collections.singleton(topicPartition), adminSupplier);
+                    if (beginningOffsets == null || beginningOffsets.isEmpty()) {
+                        LOG.error("Failed to get beginningOffsets from Kafka for topic partitions: {}.", topicPartition);
+                        return 0L;
+                    }
+                    // add value to topic level metric
+                    Long ret = endOffsets.get(topicPartition) - beginningOffsets.get(topicPartition);
+                    totalRecordsInPartitions += ret;
+                }
             }
+            return totalRecordsInPartitions;
         };
 
         metrics.put(topic + "/" + "totalSpoutLag", totalSpoutLagGauge);
@@ -115,25 +194,4 @@ public Long getValue() {
         metrics.put(topic + "/" + "totalRecordsInPartitions", totalRecordsInPartitionsGauge);
         return metrics;
     }
-
-    private class TopicMetrics {
-        long totalSpoutLag = 0L;
-        long totalEarliestTimeOffset = 0L;
-        long totalLatestTimeOffset = 0L;
-        long totalLatestEmittedOffset = 0L;
-        long totalLatestCompletedOffset = 0L;
-        long totalRecordsInPartitions = 0L;
-
-        public void incrementTotalSpoutLag(long offset) {
-            totalSpoutLag += offset;
-        }
-
-        public void incrementTotalEarliestTimeOffset(long offset) {
-            totalEarliestTimeOffset += offset;
-        }
-
-        public void incrementTotalLatestTimeOffset(long offset) {
-            totalLatestTimeOffset += offset;
-        }
-    }
 }
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetUtil.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetUtil.java
new file mode 100644
index 00000000000..c7f9fabc470
--- /dev/null
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetUtil.java
@@ -0,0 +1,92 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.storm.kafka.spout.metrics2;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.ListOffsetsResult;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.RetriableException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class KafkaOffsetUtil {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaOffsetUtil.class);
+
+    public static Map<TopicPartition, Long> getBeginningOffsets(Set<TopicPartition> topicPartitions, Supplier<Admin> adminSupplier) {
+        Admin admin = adminSupplier.get();
+        if (admin == null) {
+            LOG.error("Kafka admin object is null, returning 0.");
+            return Collections.EMPTY_MAP;
+        }
+
+        Map<TopicPartition, Long> beginningOffsets;
+        try {
+            beginningOffsets = getOffsets(admin, topicPartitions, OffsetSpec.earliest());
+        } catch (RetriableException | ExecutionException | InterruptedException e) {
+            LOG.error("Failed to get offset from Kafka for topic partitions: {}.", topicPartitions, e);
+            return Collections.EMPTY_MAP;
+        }
+        return beginningOffsets;
+    }
+
+    public static Map<TopicPartition, Long> getEndOffsets(Set<TopicPartition> topicPartitions, Supplier<Admin> adminSupplier) {
+        Admin admin = adminSupplier.get();
+        if (admin == null) {
+            LOG.error("Kafka admin object is null, returning 0.");
+            return Collections.EMPTY_MAP;
+        }
+
+        Map<TopicPartition, Long> endOffsets;
+        try {
+            endOffsets = getOffsets(admin, topicPartitions, OffsetSpec.latest());
+        } catch (RetriableException | ExecutionException | InterruptedException e) {
+            LOG.error("Failed to get offset from Kafka for topic partitions: {}.", topicPartitions, e);
+            return Collections.EMPTY_MAP;
+        }
+        return endOffsets;
+    }
+
+    public static Map<TopicPartition, Long> getOffsets(Admin admin, Set<TopicPartition> topicPartitions, OffsetSpec offsetSpec)
+            throws InterruptedException, ExecutionException {
+
+        Map<TopicPartition, OffsetSpec> offsetSpecMap = new HashMap<>();
+        for (TopicPartition topicPartition : topicPartitions) {
+            offsetSpecMap.put(topicPartition, offsetSpec);
+        }
+        Map<TopicPartition, Long> ret = new HashMap<>();
+        ListOffsetsResult listOffsetsResult = admin.listOffsets(offsetSpecMap);
+        KafkaFuture<Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo>> all = listOffsetsResult.all();
+        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionListOffsetsResultInfoMap = all.get();
+        for (Map.Entry<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> entry :
+                topicPartitionListOffsetsResultInfoMap.entrySet()) {
+            ret.put(entry.getKey(), entry.getValue().offset());
+        }
+        return ret;
+    }
+}
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutReactivationTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutReactivationTest.java
index b4cab2ace70..b7ae2824b7f 100644
--- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutReactivationTest.java
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutReactivationTest.java
@@ -29,9 +29,12 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Metric;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
@@ -65,6 +68,7 @@ public class KafkaSpoutReactivationTest {
     private final SpoutOutputCollector collector = mock(SpoutOutputCollector.class);
     private final long commitOffsetPeriodMs = 2_000;
     private Consumer<String, String> consumerSpy;
+    private Admin adminSpy;
     private KafkaSpout<String, String> spout;
     private final int maxPollRecords = 10;
 
@@ -78,9 +82,12 @@ public void prepareSpout(int messageCount, FirstPollOffsetStrategy firstPollOffs
                 .build();
         ClientFactory<String, String> clientFactory = new ClientFactoryDefault<>();
         this.consumerSpy = spy(clientFactory.createConsumer(spoutConfig.getKafkaProps()));
+        this.adminSpy = spy(clientFactory.createAdmin(spoutConfig.getKafkaProps()));
         ClientFactory<String, String> clientFactoryMock = mock(ClientFactory.class);
         when(clientFactoryMock.createConsumer(any()))
             .thenReturn(consumerSpy);
+        when(clientFactoryMock.createAdmin(any()))
+                .thenReturn(adminSpy);
         this.spout = new KafkaSpout<>(spoutConfig, clientFactoryMock, new TopicAssigner());
         SingleTopicKafkaUnitSetupHelper.populateTopicData(kafkaUnitExtension.getKafkaUnit(), SingleTopicKafkaSpoutConfiguration.TOPIC, messageCount);
         SingleTopicKafkaUnitSetupHelper.initializeSpout(spout, conf, topologyContext, collector);
@@ -147,4 +154,24 @@ public void testSpoutShouldResumeWhereItLeftOffWithEarliestStrategy() throws Exc
         //With earliest, the spout should also resume where it left off, rather than restart at the earliest offset.
         doReactivationTest(FirstPollOffsetStrategy.EARLIEST);
     }
+
+    @Test
+    public void testSpoutMustHandleGettingMetricsWhileDeactivated() throws Exception {
+        //Storm will try to get metrics from the spout even while deactivated, the spout must be able to handle this
+        prepareSpout(10, FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST);
+
+        for (int i = 0; i < 5; i++) {
+            KafkaSpoutMessageId msgId = emitOne();
+            spout.ack(msgId);
+        }
+        spout.deactivate();
+
+        Map<String, Metric> partitionsOffsetMetric = spout.getKafkaOffsetMetricManager().getTopicPartitionMetricsMap().get(new TopicPartition(SingleTopicKafkaSpoutConfiguration.TOPIC ,0)).getMetrics();
+        Long partitionLag = (Long) ((Gauge) partitionsOffsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC + "/partition_0/spoutLag")).getValue();
+        assertThat(partitionLag, is(5L));
+
+        Map<String, Metric> topicOffsetMetric = spout.getKafkaOffsetMetricManager().getTopicMetricsMap().get(SingleTopicKafkaSpoutConfiguration.TOPIC).getMetrics();
+        Long totalSpoutLag = (Long) ((Gauge) topicOffsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC + "/totalSpoutLag")).getValue();
+        assertThat(totalSpoutLag, is(5L));
+    }
 }
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/metric2/KafkaOffsetPartitionMetricsTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/metric2/KafkaOffsetPartitionMetricsTest.java
new file mode 100644
index 00000000000..ee085858e86
--- /dev/null
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/metric2/KafkaOffsetPartitionMetricsTest.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ *   or more contributor license agreements.  See the NOTICE file
+ *   distributed with this work for additional information
+ *   regarding copyright ownership.  The ASF licenses this file
+ *   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
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   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.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package org.apache.storm.kafka.spout.metric2;
+
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.*;
+
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Metric;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.ListOffsetsResult;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.storm.kafka.spout.internal.OffsetManager;
+import org.apache.storm.kafka.spout.metrics2.KafkaOffsetPartitionMetrics;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class KafkaOffsetPartitionMetricsTest {
+
+    private Set<TopicPartition> assignment;
+    private Admin admin = mock(Admin.class);
+    private HashMap<TopicPartition, OffsetManager> offsetManagers;
+    private KafkaFuture kafkaFuture = mock(KafkaFuture.class);
+
+    @BeforeEach
+    public void initializeTests() {
+        reset(admin, kafkaFuture);
+
+    }
+
+    @Test
+    public void registerMetricsGetSpoutLagAndPartitionRecords() throws ExecutionException, InterruptedException {
+
+        TopicPartition topicAPartition1 = new TopicPartition("topicA", 1);
+
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition1LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(100, System.currentTimeMillis(), Optional.empty());
+
+        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionLatestListOffsetsResultInfoMap = new HashMap<>();
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicAPartition1, topicAPartition1LatestListOffsetsResultInfo);
+
+        when(kafkaFuture.get()).thenReturn(topicPartitionLatestListOffsetsResultInfoMap);
+
+        ListOffsetsResult listOffsetsResult = mock(ListOffsetsResult.class);
+        when(listOffsetsResult.all()).thenReturn(kafkaFuture);
+
+        admin = mock(Admin.class);
+        when(admin.listOffsets(anyMap())).thenReturn(listOffsetsResult);
+
+        OffsetManager offsetManagertopicAPartition1 = mock(OffsetManager.class);
+        when(offsetManagertopicAPartition1.getCommittedOffset()).thenReturn(90L);
+
+
+        offsetManagers = new HashMap<>();
+        offsetManagers.put(topicAPartition1, offsetManagertopicAPartition1);
+
+        KafkaOffsetPartitionMetrics kafkaOffsetPartitionAndTopicMetrics = new KafkaOffsetPartitionMetrics(() -> Collections.unmodifiableMap(offsetManagers), () -> admin, topicAPartition1);
+        Map<String, Metric> result = kafkaOffsetPartitionAndTopicMetrics.getMetrics();
+        Gauge g1 = (Gauge) result.get("topicA/partition_1/spoutLag");
+        assertEquals(10L, g1.getValue());
+
+
+        //get partition records
+
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition1EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(1, System.currentTimeMillis(), Optional.empty());
+
+        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionEarliestListOffsetsResultInfoMap = new HashMap<>();
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicAPartition1, topicAPartition1EarliestListOffsetsResultInfo);
+
+        //mock consecutive calls. Each call to the recordsInPartition gauge will call kafkaFuture.get() twice
+        when(kafkaFuture.get()).thenReturn(topicPartitionLatestListOffsetsResultInfoMap, topicPartitionEarliestListOffsetsResultInfoMap);
+
+        result = kafkaOffsetPartitionAndTopicMetrics.getMetrics();
+        g1 = (Gauge) result.get("topicA/partition_1/recordsInPartition");
+        assertEquals(99L, g1.getValue());
+    }
+
+    @Test
+    public void registerMetricsGetEarliestAndLatest() throws ExecutionException, InterruptedException {
+
+        TopicPartition topicAPartition1 = new TopicPartition("topicA", 1);
+
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition1EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(1, System.currentTimeMillis(), Optional.empty());
+
+        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionEarliestListOffsetsResultInfoMap = new HashMap<>();
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicAPartition1, topicAPartition1EarliestListOffsetsResultInfo);
+
+        when(kafkaFuture.get()).thenReturn(topicPartitionEarliestListOffsetsResultInfoMap);
+
+        ListOffsetsResult listOffsetsResult = mock(ListOffsetsResult.class);
+        when(listOffsetsResult.all()).thenReturn(kafkaFuture);
+
+        admin = mock(Admin.class);
+        when(admin.listOffsets(anyMap())).thenReturn(listOffsetsResult);
+
+        OffsetManager offsetManagertopicAPartition1 = mock(OffsetManager.class);
+        when(offsetManagertopicAPartition1.getLatestEmittedOffset()).thenReturn(50L);
+        when(offsetManagertopicAPartition1.getCommittedOffset()).thenReturn(40L);
+
+
+        offsetManagers = new HashMap<>();
+        offsetManagers.put(topicAPartition1, offsetManagertopicAPartition1);
+
+        assignment = new HashSet<>();
+        assignment.add(topicAPartition1);
+
+        KafkaOffsetPartitionMetrics kafkaOffsetPartitionAndTopicMetrics = new KafkaOffsetPartitionMetrics(() -> Collections.unmodifiableMap(offsetManagers), () -> admin, topicAPartition1);
+        Map<String, Metric> result = kafkaOffsetPartitionAndTopicMetrics.getMetrics();
+        Gauge g1 = (Gauge) result.get("topicA/partition_1/earliestTimeOffset");
+
+        assertEquals(g1.getValue(), 1L);
+
+        //get the latest offsets
+
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition1LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(100, System.currentTimeMillis(), Optional.empty());
+
+
+        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionLatestListOffsetsResultInfoMap = new HashMap<>();
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicAPartition1, topicAPartition1LatestListOffsetsResultInfo);
+
+        when(kafkaFuture.get()).thenReturn(topicPartitionLatestListOffsetsResultInfoMap);
+
+        result = kafkaOffsetPartitionAndTopicMetrics.getMetrics();
+
+        g1 = (Gauge) result.get("topicA/partition_1/latestTimeOffset");
+        assertEquals(100L, g1.getValue());
+
+        g1 = (Gauge) result.get("topicA/partition_1/latestEmittedOffset");
+        assertEquals(50L, g1.getValue());
+
+
+        g1 = (Gauge) result.get("topicA/partition_1/latestCompletedOffset");
+        assertEquals(40L, g1.getValue());
+
+    }
+}
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/metric2/KafkaOffsetTopicMetricsTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/metric2/KafkaOffsetTopicMetricsTest.java
new file mode 100644
index 00000000000..26462503389
--- /dev/null
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/metric2/KafkaOffsetTopicMetricsTest.java
@@ -0,0 +1,243 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.storm.kafka.spout.metric2;
+
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Metric;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.ListOffsetsResult;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.storm.kafka.spout.internal.OffsetManager;
+import org.apache.storm.kafka.spout.metrics2.KafkaOffsetTopicMetrics;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.*;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.*;
+
+@ExtendWith(MockitoExtension.class)
+public class KafkaOffsetTopicMetricsTest {
+
+    private Set<TopicPartition> assignment;
+    private Admin admin = mock(Admin.class);
+    private HashMap<TopicPartition, OffsetManager> offsetManagers;
+    private KafkaFuture kafkaFuture = mock(KafkaFuture.class);
+
+    @BeforeEach
+    public void initializeTests() {
+        reset(admin, kafkaFuture);
+    }
+
+    @Test
+    public void registerMetricsGetSpoutLagAndPartitionRecords() throws ExecutionException, InterruptedException {
+
+        TopicPartition topicAPartition1 = new TopicPartition("topicA", 1);
+        TopicPartition topicAPartition2 = new TopicPartition("topicA", 2);
+        TopicPartition topicBPartition1 = new TopicPartition("topicB", 1);
+        TopicPartition topicBPartition2 = new TopicPartition("topicB", 2);
+
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition1LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(100, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition2LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(200, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicBPartition1LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(300, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicBPartition2LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(400, System.currentTimeMillis(), Optional.empty());
+
+        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionLatestListOffsetsResultInfoMap = new HashMap<>();
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicAPartition1, topicAPartition1LatestListOffsetsResultInfo);
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicAPartition2, topicAPartition2LatestListOffsetsResultInfo);
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicBPartition1, topicBPartition1LatestListOffsetsResultInfo);
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicBPartition2, topicBPartition2LatestListOffsetsResultInfo);
+
+        when(kafkaFuture.get()).thenReturn(topicPartitionLatestListOffsetsResultInfoMap);
+
+        ListOffsetsResult listOffsetsResult = mock(ListOffsetsResult.class);
+        when(listOffsetsResult.all()).thenReturn(kafkaFuture);
+
+        admin = mock(Admin.class);
+        when(admin.listOffsets(anyMap())).thenReturn(listOffsetsResult);
+
+        OffsetManager offsetManagertopicAPartition1 = mock(OffsetManager.class);
+        when(offsetManagertopicAPartition1.getCommittedOffset()).thenReturn(90L);
+
+        OffsetManager offsetManagertopicAPartition2 = mock(OffsetManager.class);
+        when(offsetManagertopicAPartition2.getCommittedOffset()).thenReturn(170L);
+
+        offsetManagers = new HashMap<>();
+        offsetManagers.put(topicAPartition1, offsetManagertopicAPartition1);
+        offsetManagers.put(topicAPartition2, offsetManagertopicAPartition2);
+
+        assignment = new HashSet<>();
+        assignment.add(topicAPartition1);
+        assignment.add(topicAPartition2);
+        assignment.add(topicBPartition1);
+        assignment.add(topicBPartition2);
+
+
+        KafkaOffsetTopicMetrics kafkaOffsetTopicMetricsA = new KafkaOffsetTopicMetrics("topicA", () -> Collections.unmodifiableMap(offsetManagers), () -> admin, assignment);
+        Map<String, Metric> result = kafkaOffsetTopicMetricsA.getMetrics();
+        Gauge g1 = (Gauge) result.get("topicA/totalSpoutLag");
+        assertEquals(40L, g1.getValue());
+
+        //get again the values from the Gauge. Values cannot change
+        g1 = (Gauge) result.get("topicA/totalSpoutLag");
+        assertEquals(40L, g1.getValue());
+
+        assertNull(result.get("topicB/totalSpoutLag"));
+
+        //get topic records
+
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition1EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(1, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition2EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(2, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicBPartition1EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(3, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicBPartition2EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(4, System.currentTimeMillis(), Optional.empty());
+
+        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionEarliestListOffsetsResultInfoMap = new HashMap<>();
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicAPartition1, topicAPartition1EarliestListOffsetsResultInfo);
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicAPartition2, topicAPartition2EarliestListOffsetsResultInfo);
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicBPartition1, topicBPartition1EarliestListOffsetsResultInfo);
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicBPartition2, topicBPartition2EarliestListOffsetsResultInfo);
+
+        //mock consecutive calls. Each call to the recordsInPartition gauge will call kafkaFuture.get() twice
+        when(kafkaFuture.get()).thenReturn(
+                topicPartitionLatestListOffsetsResultInfoMap, topicPartitionEarliestListOffsetsResultInfoMap,
+                topicPartitionLatestListOffsetsResultInfoMap, topicPartitionEarliestListOffsetsResultInfoMap,
+                topicPartitionLatestListOffsetsResultInfoMap, topicPartitionEarliestListOffsetsResultInfoMap,
+                topicPartitionLatestListOffsetsResultInfoMap, topicPartitionEarliestListOffsetsResultInfoMap);
+
+        Gauge gATotal = (Gauge) result.get("topicA/totalRecordsInPartitions");
+        assertEquals(297L, gATotal.getValue());
+
+        //get again the values from the Gauge. Values cannot change
+         gATotal = (Gauge) result.get("topicA/totalRecordsInPartitions");
+        assertEquals(297L, gATotal.getValue());
+
+        assertNull(result.get("topicB/totalRecordsInPartitions"));
+    }
+
+    @Test
+    public void registerMetricsGetEarliestAndLatest() throws ExecutionException, InterruptedException {
+
+        TopicPartition topicAPartition1 = new TopicPartition("topicA", 1);
+        TopicPartition topicAPartition2 = new TopicPartition("topicA", 2);
+        TopicPartition topicBPartition1 = new TopicPartition("topicB", 1);
+        TopicPartition topicBPartition2 = new TopicPartition("topicB", 2);
+
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition1EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(1, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition2EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(1, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicBPartition1EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(1, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicBPartition2EarliestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(1, System.currentTimeMillis(), Optional.empty());
+
+        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionEarliestListOffsetsResultInfoMap = new HashMap<>();
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicAPartition1, topicAPartition1EarliestListOffsetsResultInfo);
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicAPartition2, topicAPartition2EarliestListOffsetsResultInfo);
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicBPartition1, topicBPartition1EarliestListOffsetsResultInfo);
+        topicPartitionEarliestListOffsetsResultInfoMap.put(topicBPartition2, topicBPartition2EarliestListOffsetsResultInfo);
+
+        when(kafkaFuture.get()).thenReturn(topicPartitionEarliestListOffsetsResultInfoMap);
+
+        ListOffsetsResult listOffsetsResult = mock(ListOffsetsResult.class);
+        when(listOffsetsResult.all()).thenReturn(kafkaFuture);
+
+        admin = mock(Admin.class);
+        when(admin.listOffsets(anyMap())).thenReturn(listOffsetsResult);
+
+        OffsetManager offsetManagertopicAPartition1 = mock(OffsetManager.class);
+        when(offsetManagertopicAPartition1.getLatestEmittedOffset()).thenReturn(50L);
+        when(offsetManagertopicAPartition1.getCommittedOffset()).thenReturn(40L);
+
+        OffsetManager offsetManagertopicAPartition2 = mock(OffsetManager.class);
+        when(offsetManagertopicAPartition2.getLatestEmittedOffset()).thenReturn(100L);
+        when(offsetManagertopicAPartition2.getCommittedOffset()).thenReturn(90L);
+
+
+        offsetManagers = new HashMap<>();
+        offsetManagers.put(topicAPartition1, offsetManagertopicAPartition1);
+        offsetManagers.put(topicAPartition2, offsetManagertopicAPartition2);
+
+        assignment = new HashSet<>();
+        assignment.add(topicAPartition1);
+        assignment.add(topicAPartition2);
+        assignment.add(topicBPartition1);
+        assignment.add(topicBPartition2);
+
+        KafkaOffsetTopicMetrics kafkaOffsetPartitionAndTopicMetrics = new KafkaOffsetTopicMetrics("topicA",() -> Collections.unmodifiableMap(offsetManagers), () -> admin, assignment);
+        Map<String, Metric> result = kafkaOffsetPartitionAndTopicMetrics.getMetrics();
+
+        Gauge gATotal = (Gauge) result.get("topicA/totalEarliestTimeOffset");
+        assertEquals(2L, gATotal.getValue());
+        assertNull(result.get("topicB/totalEarliestTimeOffset"));
+
+        //get the metrics a second time. Values should be the same
+
+        gATotal = (Gauge) result.get("topicA/totalEarliestTimeOffset");
+        assertEquals(2L, gATotal.getValue());
+        assertNull(result.get("topicB/totalEarliestTimeOffset"));
+
+        //get the latest offsets
+
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition1LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(100, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicAPartition2LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(200, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicBPartition1LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(300, System.currentTimeMillis(), Optional.empty());
+        ListOffsetsResult.ListOffsetsResultInfo topicBPartition2LatestListOffsetsResultInfo = new ListOffsetsResult.ListOffsetsResultInfo(400, System.currentTimeMillis(), Optional.empty());
+
+        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> topicPartitionLatestListOffsetsResultInfoMap = new HashMap<>();
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicAPartition1, topicAPartition1LatestListOffsetsResultInfo);
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicAPartition2, topicAPartition2LatestListOffsetsResultInfo);
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicBPartition1, topicBPartition1LatestListOffsetsResultInfo);
+        topicPartitionLatestListOffsetsResultInfoMap.put(topicBPartition2, topicBPartition2LatestListOffsetsResultInfo);
+
+        when(kafkaFuture.get()).thenReturn(topicPartitionLatestListOffsetsResultInfoMap);
+
+        gATotal = (Gauge) result.get("topicA/totalLatestTimeOffset");
+        assertEquals(300L, gATotal.getValue());
+        assertNull((result.get("topicB/totalLatestTimeOffset")));
+
+        gATotal = (Gauge) result.get("topicA/totalLatestEmittedOffset");
+        assertEquals(150L, gATotal.getValue());
+        assertNull( result.get("topicB/totalLatestEmittedOffset"));
+
+
+        gATotal = (Gauge) result.get("topicA/totalLatestCompletedOffset");
+        assertEquals(130L, gATotal.getValue());
+        assertNull(result.get("topiBA/totalLatestCompletedOffset"));
+
+        //get the metrics a second time. Values should be the same
+
+        gATotal = (Gauge) result.get("topicA/totalLatestTimeOffset");
+        assertEquals(300L, gATotal.getValue());
+        assertNull((result.get("topicB/totalLatestTimeOffset")));
+
+        gATotal = (Gauge) result.get("topicA/totalLatestEmittedOffset");
+        assertEquals(150L, gATotal.getValue());
+        assertNull( result.get("topicB/totalLatestEmittedOffset"));
+
+
+        gATotal = (Gauge) result.get("topicA/totalLatestCompletedOffset");
+        assertEquals(130L, gATotal.getValue());
+        assertNull(result.get("topiBA/totalLatestCompletedOffset"));
+
+    }
+}