Add Time based metric exipration. (#1996)

Fixes #1956

### Motivation
FastThreadLocal wont cleanup until thread exit. so we can just check if
the stats is not recorded a long time if expired just remove from
threadlocal


### Modifications
add time check logic if long time have no record just remove the stats.

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/DataSketchesOpStatsLogger.java

@@ -101,15 +101,15 @@ public void clear() {
         throw new UnsupportedOperationException();
     }
 
-    public void rotateLatencyCollection() {
+    public void rotateLatencyCollection(long expiredTimeSeconds) {
         // Swap current with replacement
         ThreadLocalAccessor local = current;
         current = replacement;
         replacement = local;
 
         final DoublesUnion aggregateSuccess = new DoublesUnionBuilder().build();
         final DoublesUnion aggregateFail = new DoublesUnionBuilder().build();
-        local.record(aggregateSuccess, aggregateFail);
+        local.recordAndCheckStatsExpire(aggregateSuccess, aggregateFail, expiredTimeSeconds);
         successResult = aggregateSuccess.getResultAndReset();
         failResult = aggregateFail.getResultAndReset();
     }

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/LocalData.java

@@ -23,6 +23,7 @@ public class LocalData {
     private final DoublesSketch successSketch = new DoublesSketchBuilder().build();
     private final DoublesSketch failSketch = new DoublesSketchBuilder().build();
     private final StampedLock lock = new StampedLock();
+    private volatile long lastHasRecordTime = System.currentTimeMillis();
 
     public void updateSuccessSketch(double value) {
         long stamp = lock.readLock();
@@ -42,9 +43,18 @@ public void updateFailedSketch(double value) {
         }
     }
 
+    public long lastHasRecordTime() {
+        return lastHasRecordTime;
+    }
+
     public void record(DoublesUnion aggregateSuccess, DoublesUnion aggregateFail) {
         long stamp = lock.writeLock();
         try {
+            boolean isEmpty = successSketch.isEmpty() && failSketch.isEmpty();
+            if (!isEmpty) {
+                lastHasRecordTime = System.currentTimeMillis();
+            }
+
             aggregateSuccess.update(successSketch);
             successSketch.reset();
             aggregateFail.update(failSketch);

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusMetricsProvider.java

@@ -38,6 +38,12 @@ public class PrometheusMetricsProvider implements PrometheusRawMetricsProvider {
     public static final String PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS = "prometheusStatsLatencyRolloverSeconds";
     public static final int DEFAULT_PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS = 60;
 
+    public static final String PROMETHEUS_STATS_EXPIRED_SECONDS = "prometheusStatsExpiredSeconds";
+
+    public static final long DEFAULT_PROMETHEUS_STATS_EXPIRED_SECONDS = TimeUnit.HOURS.toSeconds(1);
+
+    private long expiredTimeSeconds;
+
     private static final String KOP_PROMETHEUS_STATS_CLUSTER = "cluster";
     private final Map<String, String> defaultStatsLoggerLabels = new HashMap<>();
 
@@ -65,11 +71,14 @@ public void start(Configuration conf) {
         int latencyRolloverSeconds = conf.getInt(PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS,
                 DEFAULT_PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS);
 
+        expiredTimeSeconds = conf.getLong(PROMETHEUS_STATS_EXPIRED_SECONDS,
+                DEFAULT_PROMETHEUS_STATS_EXPIRED_SECONDS);
+
         defaultStatsLoggerLabels.putIfAbsent(KOP_PROMETHEUS_STATS_CLUSTER,
                 conf.getString(KOP_PROMETHEUS_STATS_CLUSTER));
 
         executor.scheduleAtFixedRate(() -> {
-            rotateLatencyCollection();
+            rotateLatencyCollectionAndExpire(expiredTimeSeconds);
         }, 1, latencyRolloverSeconds, TimeUnit.SECONDS);
 
     }
@@ -103,9 +112,9 @@ public String getStatsName(String... statsComponents) {
     }
 
     @VisibleForTesting
-    void rotateLatencyCollection() {
+    void rotateLatencyCollectionAndExpire(long expiredTimeSeconds) {
         opStats.forEach((name, metric) -> {
-            metric.rotateLatencyCollection();
+            metric.rotateLatencyCollection(expiredTimeSeconds);
         });
     }
 }

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/ThreadLocalAccessor.java

@@ -36,8 +36,20 @@ protected void onRemoval(LocalData value) {
         }
     };
 
-    public void record(DoublesUnion aggregateSuccess, DoublesUnion aggregateFail) {
-        map.keySet().forEach(key -> key.record(aggregateSuccess, aggregateFail));
+    public void recordAndCheckStatsExpire(DoublesUnion aggregateSuccess,
+                                          DoublesUnion aggregateFail,
+                                          long expireTimeMs) {
+        long currentTime = System.currentTimeMillis();
+
+        map.keySet().forEach(key -> {
+            // update stats
+            key.record(aggregateSuccess, aggregateFail);
+
+            // check if record expired.
+            if (currentTime - key.lastHasRecordTime() > expireTimeMs) {
+                map.remove(key);
+            }
+        });
     }
 
     public LocalData getLocalData() {