Add state transition latency metrics for Helix participant (#3283)

* Add state transition latency metrics for Helix participant

Add metrics to track BOOTSTRAP→STANDBY duration, OFFLINE→BOOTSTRAP duration,
bootstrap failures, and max time-in-BOOTSTRAP. These fill a gap where no
latency instrumentation exists today (Helix's native StateTransitionStatMonitor
is denylisted from InGraph due to high cardinality).

New metrics:
- bootstrapToStandbyDurationMs: histogram of successful bootstrap durations
- offlineToBootstrapDurationMs: histogram of offline→bootstrap transition time
- bootstrapFailureCount: counter of partitions that failed during bootstrap
- maxTimeInBootstrapMs: gauge of the longest currently-bootstrapping partition

Motivation: INC-15835 (Helix cascading OOM) showed that we lacked visibility
into how long partitions spend in BOOTSTRAP during rebalance storms.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* Remove 4h stale bootstrap eviction — map bounded by partition count per instance (~1.2k max)

---------

Co-authored-by: Cris Liao <crliao@linkedin.com>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: 3 people

ambry-clustermap/src/main/java/com/github/ambry/clustermap/HelixParticipant.java

@@ -918,6 +918,7 @@ public void onPartitionBecomeBootstrapFromPreBootstrap(String partitionName) {
 
   @Override
   public void onPartitionBecomeBootstrapFromOffline(String partitionName) {
+    long transitionStartMs = System.currentTimeMillis();
     try {
       if (this.blockStateTransitionLatch != null && this.blockStateTransitionLatch.getCount() > 0) {
         logger.info("Bootstrapping is waiting for blockStateTransitionLatch...");
@@ -963,10 +964,13 @@ public void onPartitionBecomeBootstrapFromOffline(String partitionName) {
       logger.error("Waiting for state transition to be unblocked was interrupted", e);
     } catch (Exception e) {
       localPartitionAndState.put(partitionName, ReplicaState.ERROR);
+      participantMetrics.recordBootstrapFailure(partitionName);
       throw e;
     }
     logger.info("Before setting partition {} to bootstrap", partitionName);
     localPartitionAndState.put(partitionName, ReplicaState.BOOTSTRAP);
+    participantMetrics.recordOfflineToBootstrapDuration(System.currentTimeMillis() - transitionStartMs);
+    participantMetrics.recordBootstrapStart(partitionName);
     participantMetrics.decStateTransitionMetric(partitionName, ReplicaState.OFFLINE, ReplicaState.BOOTSTRAP);
   }
 
@@ -992,13 +996,16 @@ public void onPartitionBecomeStandbyFromBootstrap(String partitionName) {
     } catch (InterruptedException e) {
       logger.error("Bootstrap was interrupted on partition {}", partitionName);
       localPartitionAndState.put(partitionName, ReplicaState.ERROR);
+      participantMetrics.recordBootstrapFailure(partitionName);
       throw new StateTransitionException("Bootstrap failed or was interrupted", BootstrapFailure);
     } catch (StateTransitionException e) {
       logger.error("Bootstrap didn't complete on partition {}", partitionName, e);
       localPartitionAndState.put(partitionName, ReplicaState.ERROR);
+      participantMetrics.recordBootstrapFailure(partitionName);
       throw e;
     }
     localPartitionAndState.put(partitionName, ReplicaState.STANDBY);
+    participantMetrics.recordBootstrapComplete(partitionName);
     participantMetrics.decStateTransitionMetric(partitionName, ReplicaState.BOOTSTRAP, ReplicaState.STANDBY);
   }
 

ambry-clustermap/src/main/java/com/github/ambry/clustermap/HelixParticipantMetrics.java

@@ -15,10 +15,12 @@
 
 import com.codahale.metrics.Counter;
 import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Histogram;
 import com.codahale.metrics.MetricRegistry;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 
 /**
@@ -38,6 +40,16 @@ class HelixParticipantMetrics {
 
   public final Counter updateDiskCapacityCounter;
 
+  // --- State Transition Latency Metrics ---
+  // Tracks when each partition entered BOOTSTRAP state (epoch ms)
+  private final ConcurrentHashMap<String, Long> bootstrapStartTimeMs = new ConcurrentHashMap<>();
+  // Histogram of successful BOOTSTRAP→STANDBY durations in milliseconds
+  final Histogram bootstrapToStandbyDurationMs;
+  // Histogram of OFFLINE→BOOTSTRAP transition durations in milliseconds
+  final Histogram offlineToBootstrapDurationMs;
+  // Counter for bootstrap failures (partition went to ERROR from BOOTSTRAP)
+  final Counter bootstrapFailureCount;
+
   final Map<String, Counter> partitionTransitionToCount;
 
   HelixParticipantMetrics(MetricRegistry metricRegistry, String zkConnectStr,
@@ -72,6 +84,17 @@ class HelixParticipantMetrics {
     updateDiskCapacityCounter =
         metricRegistry.counter(MetricRegistry.name(HelixParticipant.class, "updateDiskCapacityCount"));
     partitionTransitionToCount = new HashMap<>();
+
+    // State transition latency metrics
+    bootstrapToStandbyDurationMs = metricRegistry.histogram(
+        MetricRegistry.name(HelixParticipant.class, "bootstrapToStandbyDurationMs" + zkSuffix));
+    offlineToBootstrapDurationMs = metricRegistry.histogram(
+        MetricRegistry.name(HelixParticipant.class, "offlineToBootstrapDurationMs" + zkSuffix));
+    bootstrapFailureCount = metricRegistry.counter(
+        MetricRegistry.name(HelixParticipant.class, "bootstrapFailureCount" + zkSuffix));
+    Gauge<Long> maxTimeInBootstrap = this::computeMaxTimeInBootstrap;
+    registry.gauge(MetricRegistry.name(HelixParticipant.class, "maxTimeInBootstrapMs" + zkSuffix),
+        () -> maxTimeInBootstrap);
   }
 
   /**
@@ -95,6 +118,54 @@ private int getReplicaCountInState(ReplicaState state) {
     return replicaCountByState.get(state);
   }
 
+  /**
+   * Record that a partition has entered BOOTSTRAP state.
+   * @param partitionName the partition that entered BOOTSTRAP
+   */
+  void recordBootstrapStart(String partitionName) {
+    bootstrapStartTimeMs.put(partitionName, System.currentTimeMillis());
+  }
+
+  /**
+   * Record that a partition has completed BOOTSTRAP→STANDBY transition successfully.
+   * @param partitionName the partition that reached STANDBY
+   */
+  void recordBootstrapComplete(String partitionName) {
+    Long startTime = bootstrapStartTimeMs.remove(partitionName);
+    if (startTime != null) {
+      bootstrapToStandbyDurationMs.update(System.currentTimeMillis() - startTime);
+    }
+  }
+
+  /**
+   * Record that a partition failed during BOOTSTRAP (went to ERROR).
+   * @param partitionName the partition that failed
+   */
+  void recordBootstrapFailure(String partitionName) {
+    bootstrapStartTimeMs.remove(partitionName);
+    bootstrapFailureCount.inc();
+  }
+
+  /**
+   * Record the duration of an OFFLINE→BOOTSTRAP transition.
+   * @param durationMs time in milliseconds the transition took
+   */
+  void recordOfflineToBootstrapDuration(long durationMs) {
+    offlineToBootstrapDurationMs.update(durationMs);
+  }
+
+  /**
+   * Compute the maximum time any partition has been in BOOTSTRAP state.
+   */
+  private long computeMaxTimeInBootstrap() {
+    long now = System.currentTimeMillis();
+    long maxDuration = 0;
+    for (Map.Entry<String, Long> entry : bootstrapStartTimeMs.entrySet()) {
+      maxDuration = Math.max(maxDuration, now - entry.getValue());
+    }
+    return maxDuration;
+  }
+
   /**
    * Creates and increments the metric object for given partition's state transition
    * @param partitionName partition name