[server] Introduce underReplicatedBuckets metrics (apache#798)

Author: libenchao

fluss-common/src/main/java/com/alibaba/fluss/metrics/MetricNames.java

@@ -97,6 +97,7 @@ public class MetricNames {
     // --------------------------------------------------------------------------------------------
 
     // for replica
+    public static final String UNDER_REPLICATED = "underReplicated";
     public static final String IN_SYNC_REPLICAS = "inSyncReplicasCount";
     public static final String UNDER_MIN_ISR = "underMinIsr";
     public static final String AT_MIN_ISR = "atMinIsr";

fluss-server/src/main/java/com/alibaba/fluss/server/replica/Replica.java

@@ -244,6 +244,7 @@ public Replica(
     }
 
     private void registerMetrics() {
+        bucketMetricGroup.gauge(MetricNames.UNDER_REPLICATED, () -> isUnderReplicated() ? 1 : 0);
         bucketMetricGroup.gauge(
                 MetricNames.IN_SYNC_REPLICAS, () -> isLeader() ? isrState.isr().size() : 0);
         bucketMetricGroup.gauge(MetricNames.UNDER_MIN_ISR, () -> isUnderMinIsr() ? 1 : 0);
@@ -1731,6 +1732,11 @@ private void validateInSyncReplicaSize(int requiredAcks) {
         }
     }
 
+    private boolean isUnderReplicated() {
+        // is leader and isr size less than numReplicas
+        return isLeader() && isrState.isr().size() < tableConfig.getReplicationFactor();
+    }
+
     private boolean isUnderMinIsr() {
         return isLeader() && isrState.isr().size() < minInSyncReplicas;
     }