[fix][broker]pulsar_ml_reads_inflight_bytes and pulsar_ml_reads_available_inflight_bytes are 0 at the same time (#25105)

Author: codelipenghui

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/InflightReadsLimiter.java

@@ -42,7 +42,8 @@ public class InflightReadsLimiter implements AutoCloseable {
 
     @PulsarDeprecatedMetric(newMetricName = INFLIGHT_READS_LIMITER_LIMIT_METRIC_NAME)
     @Deprecated
-    private static final Gauge PULSAR_ML_READS_BUFFER_SIZE = Gauge
+    @VisibleForTesting
+    static final Gauge PULSAR_ML_READS_BUFFER_SIZE = Gauge
             .build()
             .name("pulsar_ml_reads_inflight_bytes")
             .help("Estimated number of bytes retained by data read from storage or cache")
@@ -55,7 +56,8 @@ public class InflightReadsLimiter implements AutoCloseable {
 
     @PulsarDeprecatedMetric(newMetricName = INFLIGHT_READS_LIMITER_USAGE_METRIC_NAME)
     @Deprecated
-    private static final Gauge PULSAR_ML_READS_AVAILABLE_BUFFER_SIZE = Gauge
+    @VisibleForTesting
+    static final Gauge PULSAR_ML_READS_AVAILABLE_BUFFER_SIZE = Gauge
             .build()
             .name("pulsar_ml_reads_available_inflight_bytes")
             .help("Available space for inflight data read from storage or cache")
@@ -87,6 +89,7 @@ public InflightReadsLimiter(long maxReadsInFlightSize, int maxReadsInFlightAcqui
         if (maxReadsInFlightSize > 0) {
             enabled = true;
             this.queuedHandles = new ArrayDeque<>();
+            updateMetrics();
         } else {
             enabled = false;
             this.queuedHandles = null;

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/InflightReadsLimiterTest.java

@@ -538,6 +538,18 @@ public void testAcquireExceedingMaxReadsWhenAllPermitsAvailable() throws Excepti
                 .isEqualTo(maxReadsInFlightSize);
     }
 
+    @Test
+    public void testPrometheusMetrics() throws Exception {
+        long maxReadsInFlightSize = 100;
+        @Cleanup
+        InflightReadsLimiter limiter = new InflightReadsLimiter(maxReadsInFlightSize, ACQUIRE_QUEUE_SIZE,
+                ACQUIRE_TIMEOUT_MILLIS, mock(ScheduledExecutorService.class), OpenTelemetry.noop());
+
+        Assertions.assertThat(limiter.PULSAR_ML_READS_BUFFER_SIZE.get()).isZero();
+        Assertions.assertThat(limiter.PULSAR_ML_READS_AVAILABLE_BUFFER_SIZE.get())
+                .isEqualTo(maxReadsInFlightSize);
+    }
+
     private Pair<OpenTelemetrySdk, InMemoryMetricReader> buildOpenTelemetryAndReader() {
         var metricReader = InMemoryMetricReader.create();
         var openTelemetry = AutoConfiguredOpenTelemetrySdk.builder()