feat(metrics): skip reporting metrics delay for newly created partiti…

…on (#2028)

give some exemption time (60s) for newly created partition to receive metrics before reporting it as out-of-sync to prevent false alert

Signed-off-by: Shichao Nie <niesc@automq.com>

core/src/main/java/kafka/autobalancer/model/AbstractInstanceUpdater.java

@@ -31,6 +31,15 @@ public abstract class AbstractInstanceUpdater {
     protected Map<Byte, Samples> metricSampleMap = new HashMap<>();
     protected long lastUpdateTimestamp = 0L;
     protected MetricVersion metricVersion = defaultVersion();
+    private final long createTimestamp;
+
+    public AbstractInstanceUpdater() {
+        this.createTimestamp = System.currentTimeMillis();
+    }
+
+    public long createTimestamp() {
+        return createTimestamp;
+    }
 
     public boolean update(Iterable<Map.Entry<Byte, Double>> metricsMap, long time) {
         lock.lock();

core/src/main/java/kafka/autobalancer/model/ClusterModel.java

@@ -32,7 +32,7 @@ public class ClusterModel {
     protected final Logger logger;
     private static final String DEFAULT_RACK_ID = "rack_default";
     private static final long DEFAULT_MAX_TOLERATED_METRICS_DELAY_MS = 60000L;
-
+    private static final long DEFAULT_METRICS_DELAY_EXEMPTION_TIME_MS = 60000L;
     /*
      * Guard the access on cluster structure (read/add/remove for brokers, replicas)
      */
@@ -78,6 +78,10 @@ Map<Integer, Long> calculateBrokerLatestMetricsTime() {
                 Map<TopicPartition, TopicPartitionReplicaUpdater> replicaMap = entry.getValue();
                 for (Map.Entry<TopicPartition, TopicPartitionReplicaUpdater> tpEntry : replicaMap.entrySet()) {
                     TopicPartitionReplicaUpdater replicaUpdater = tpEntry.getValue();
+                    if (System.currentTimeMillis() - replicaUpdater.createTimestamp() <= DEFAULT_METRICS_DELAY_EXEMPTION_TIME_MS) {
+                        // exempt the newly created partition
+                        continue;
+                    }
                     metricsTimeMap.put(brokerId, Math.min(metricsTimeMap.get(brokerId), replicaUpdater.getLastUpdateTimestamp()));
                 }
             }

core/src/test/java/kafka/autobalancer/model/ClusterModelTest.java

@@ -392,7 +392,7 @@ public void testMetricsTime() {
 
         Map<Integer, Long> metricsTimeMap = clusterModel.calculateBrokerLatestMetricsTime();
         Assertions.assertEquals(1, metricsTimeMap.size());
-        Assertions.assertEquals(now - 2000, metricsTimeMap.get(brokerId));
+        Assertions.assertEquals(now, metricsTimeMap.get(brokerId));
     }
 
     @Test