linkedin · LZD-PratyushBhatt · Jun 18, 2026 · Jun 23, 2026 · Jul 3, 2026 · Jul 3, 2026
@@ -974,6 +974,15 @@ private void handleEvent(ClusterEvent event, BaseControllerDataProvider dataProv
         _clusterStatusMonitor
             .updateClusterEventDuration(ClusterEventMonitor.PhaseName.TotalProcessed.name(),
                 _lastPipelineEndTimestamp - startTime);
+        // Report DEFAULT-pipeline progress so ControllerPipelineStalledGauge can tell a wedged
+        // controller (queue not draining) apart from an idle or busy-but-progressing one.
+        _clusterStatusMonitor.setLastPipelineEndTimestamp(_lastPipelineEndTimestamp);
+        // Keep the stall threshold in sync with ClusterConfig so it can be tuned without a redeploy.
+        ClusterConfig stallThresholdConfig = dataProvider.getClusterConfig();
+        if (stallThresholdConfig != null) {
+          _clusterStatusMonitor.setPipelineStallThresholdMs(
+              stallThresholdConfig.getControllerPipelineStallThresholdMs());
+        }
         if (shouldCountTopologyEventAsProcessed(rebalanceFail, dataProvider)) {
           _clusterStatusMonitor.incrementTopologyChangeEventProcessed(event.getEventType());
         }
@@ -1342,6 +1351,21 @@ private void enqueueEvent(ClusterEventBlockingQueue queue, ClusterEvent event) {
       return;
     }
     queue.put(event);
+    if (queue == _eventQueue) {
+      updateControllerEventQueueSizeGauge();
+    }
+  }
+
+  /**
+   * Publish the current DEFAULT cluster-event pipeline backlog to the per-cluster monitor. Invoked
+   * on both the enqueue side (ZK-callback / periodic-rebalance threads) and the dequeue side (the
+   * pipeline thread) so the gauge climbs when events pile up faster than they are drained, which
+   * surfaces a controller that still holds leadership but has stopped processing ("zombie leader").
+   */
+  private void updateControllerEventQueueSizeGauge() {
+    if (_isMonitoring && _clusterStatusMonitor != null && _eventQueue != null) {
+      _clusterStatusMonitor.setControllerEventQueueSizeGauge(_eventQueue.size());
+    }
   }
 
   @Override
@@ -1523,6 +1547,11 @@ private void enableClusterStatusMonitor(boolean enable) {
             _resourceControlDataProvider.clearMonitoringRecords();
           }
           _clusterStatusMonitor.active();
+          // Seed the pipeline-progress baseline at monitoring-enable (leadership acquisition) so a
+          // controller that wedges before completing its very first pipeline run is still caught:
+          // with a 0 baseline the stalled gauge cannot fire. Cold-start slowness that briefly reads
+          // as stalled is covered by the EKG warm-up window and the configurable stall threshold.
+          _clusterStatusMonitor.setLastPipelineEndTimestamp(System.currentTimeMillis());
         } else {
           logger.info("Disable clusterStatusMonitor for cluster " + _clusterName);
           // Reset will be done if (_isMonitoring = false) later, no matter if the state is changed or not.
@@ -1579,6 +1608,9 @@ public void run() {
       while (!isInterrupted()) {
         try {
           ClusterEvent newClusterEvent = _eventBlockingQueue.take();
+          if (_eventBlockingQueue == _eventQueue) {
+            updateControllerEventQueueSizeGauge();
+          }
           String threadName = String.format(
               "HelixController-pipeline-%s-(%s)", _processorName, newClusterEvent.getEventId());
           this.setName(threadName);

@@ -67,6 +67,9 @@ public enum ClusterConfigProperty {
     TOP_STATE_HANDOFF_DURATION_THRESHOLD,
     RESOURCE_PRIORITY_FIELD,
     REBALANCE_TIMER_PERIOD,
+    // Time in ms after which a non-empty controller event queue that has not completed a pipeline
+    // run is treated as a wedged ("zombie leader") controller by ControllerPipelineStalledGauge.
+    CONTROLLER_PIPELINE_STALL_THRESHOLD_MS,
     MAX_CONCURRENT_TASK_PER_INSTANCE,
 
     // The following concerns maintenance mode
@@ -238,6 +241,8 @@ public enum GlobalRebalancePreferenceKey {
   private final static int DEFAULT_VIEW_CLUSTER_REFRESH_PERIOD = 30;
   private final static long DEFAULT_LAST_ON_DEMAND_REBALANCE_TIMESTAMP = -1L;
   private final static long DEFAULT_TOP_STATE_HANDOFF_DURATION_THRESHOLD = 300000L; // 5 minutes
+  // Default for CONTROLLER_PIPELINE_STALL_THRESHOLD_MS when unset.
+  private final static long DEFAULT_CONTROLLER_PIPELINE_STALL_THRESHOLD_MS = 5000L; // 5 seconds
 
   /**
    * Instantiate for a specific cluster
@@ -833,6 +838,27 @@ public long getTopStateHandoffDurationThreshold() {
         DEFAULT_TOP_STATE_HANDOFF_DURATION_THRESHOLD);
   }
 
+  /**
+   * Set the wedged-controller stall threshold: a non-empty controller event queue that has not
+   * completed a pipeline run within this many ms is reported as stalled by
+   * ControllerPipelineStalledGauge.
+   * @param thresholdMs threshold in milliseconds
+   */
+  public void setControllerPipelineStallThresholdMs(long thresholdMs) {
+    _record.setLongField(ClusterConfigProperty.CONTROLLER_PIPELINE_STALL_THRESHOLD_MS.name(),
+        thresholdMs);
+  }
+
+  /**
+   * @return the wedged-controller stall threshold in ms, defaulting to
+   *         {@value #DEFAULT_CONTROLLER_PIPELINE_STALL_THRESHOLD_MS} when unset.
+   */
+  public long getControllerPipelineStallThresholdMs() {
+    return _record.getLongField(
+        ClusterConfigProperty.CONTROLLER_PIPELINE_STALL_THRESHOLD_MS.name(),
+        DEFAULT_CONTROLLER_PIPELINE_STALL_THRESHOLD_MS);
+  }
+
   /**
    * Set cluster level state transition time out
    * @param stateTransitionTimeoutConfig

@@ -94,6 +94,21 @@ public class ClusterStatusMonitor implements ClusterStatusMonitorMBean {
   private AtomicLong _rebalanceFailureCount = new AtomicLong(0L);
   private AtomicLong _continuousResourceRebalanceFailureCount = new AtomicLong(0L);
   private AtomicLong _continuousTaskRebalanceFailureCount = new AtomicLong(0L);
+  // DEFAULT controller cluster-event pipeline backlog. Near 0 on a healthy controller (the queue
+  // dedups by event type); climbs when the controller still holds leadership but stops draining
+  // events, surfacing the "zombie leader" failure mode.
+  private AtomicLong _controllerEventQueueSizeGauge = new AtomicLong(0L);
+  // Wall-clock time (ms) of the last completed DEFAULT controller pipeline run, reported by
+  // GenericHelixController. Paired with the queue-size gauge to derive
+  // ControllerPipelineStalledGauge. 0 until the first pipeline completes (treated as "no data").
+  private AtomicLong _lastPipelineEndTimestamp = new AtomicLong(0L);
+  // Stall threshold (ms): a non-empty event queue whose pipeline has not completed within this many
+  // ms is treated as a wedged ("zombie leader") controller. Sourced from ClusterConfig
+  // (CONTROLLER_PIPELINE_STALL_THRESHOLD_MS) and pushed by GenericHelixController each pipeline run;
+  // uses the default until first reported.
+  private static final long DEFAULT_PIPELINE_STALL_THRESHOLD_MS = 5000L;
+  private AtomicLong _pipelineStallThresholdMs =
+      new AtomicLong(DEFAULT_PIPELINE_STALL_THRESHOLD_MS);
 
   // WAGED per-FailureCategory counters. Populated in the constructor with a zero AtomicLong per
   // enum value so reads on never-incremented categories return 0 instead of NPE.
@@ -918,6 +933,15 @@ public void reset() {
       _wagedInternalFailure = false;
       _wagedBaselineComputeFailing = false;
       _wagedRebalanceOverwriteFailing = false;
+      // Zero the DEFAULT controller-event pipeline backlog gauge on leadership change, for the
+      // same reason as the counters above: the ClusterStatusMonitor instance is reused across
+      // leadership periods, so a stale depth from a prior leader must not be re-reported by the
+      // re-registered bean after re-election (it would otherwise persist until the next
+      // enqueue/dequeue refreshes it).
+      _controllerEventQueueSizeGauge.set(0L);
+      // Reset the pipeline-progress timestamp for the same reason: a stale value from a prior
+      // leadership period must not make the re-registered bean report a spurious stall.
+      _lastPipelineEndTimestamp.set(0L);
     } catch (Exception e) {
       LOG.error("Fail to reset ClusterStatusMonitor, cluster: " + _clusterName, e);
     }
@@ -1467,6 +1491,36 @@ public void reportContinuousTaskRebalanceFailureCount(long newValue) {
     _continuousTaskRebalanceFailureCount.set(newValue);
   }
 
+  /**
+   * Surface the DEFAULT controller cluster-event pipeline backlog as a JMX gauge. A healthy
+   * controller drains events quickly so this stays near 0 (the queue dedups by event type); a
+   * wedged controller that still holds leadership but stops processing lets it climb, making the
+   * "zombie leader" failure mode detectable.
+   */
+  public void setControllerEventQueueSizeGauge(long size) {
+    _controllerEventQueueSizeGauge.set(size);
+  }
+
+  /**
+   * Report the wall-clock time (ms) of the most recent completed DEFAULT controller pipeline run.
+   * Consumed by {@link #getControllerPipelineStalledGauge()} to tell a wedged controller (queue
+   * not draining) apart from a healthy idle or busy-but-progressing one.
+   */
+  public void setLastPipelineEndTimestamp(long timestampMs) {
+    _lastPipelineEndTimestamp.set(timestampMs);
+  }
+
+  /**
+   * Set the wedged-controller stall threshold (ms), sourced from ClusterConfig
+   * (CONTROLLER_PIPELINE_STALL_THRESHOLD_MS) by GenericHelixController. Non-positive values are
+   * ignored so a misconfiguration cannot silently disable {@link #getControllerPipelineStalledGauge()}.
+   */
+  public void setPipelineStallThresholdMs(long thresholdMs) {
+    if (thresholdMs > 0) {
+      _pipelineStallThresholdMs.set(thresholdMs);
+    }
+  }
+
   @Override
   public long getRebalanceFailureCounter() {
     return _rebalanceFailureCount.get();
@@ -1482,6 +1536,26 @@ public long getContinuousTaskRebalanceFailureCount() {
     return _continuousTaskRebalanceFailureCount.get();
   }
 
+  @Override
+  public long getControllerEventQueueSizeGauge() {
+    return _controllerEventQueueSizeGauge.get();
+  }
+
+  @Override
+  public long getControllerPipelineStalledGauge() {
+    long lastEnd = _lastPipelineEndTimestamp.get();
+    // 0 = not stalled: either the queue is empty (idle is healthy) or no pipeline has completed yet
+    // (no baseline). 1 = a non-empty queue whose DEFAULT pipeline has not completed within the
+    // stall threshold, i.e. the controller holds events but is not draining them (wedged / "zombie
+    // leader"). Computed lazily on read so it stays correct even if the pipeline thread is dead and
+    // no setter runs.
+    if (_controllerEventQueueSizeGauge.get() > 0 && lastEnd > 0
+        && (System.currentTimeMillis() - lastEnd) > _pipelineStallThresholdMs.get()) {
+      return 1L;
+    }
+    return 0L;
+  }
+
   @Override
   public long getWagedCustomerActionableFailureCounter() {
     return _wagedCustomerActionableFailureCount.get();

@@ -97,6 +97,31 @@ public interface ClusterStatusMonitorMBean extends SensorNameProvider {
    */
   long getContinuousTaskRebalanceFailureCount();
 
+  /**
+   * Backlog of the DEFAULT controller cluster-event pipeline (events enqueued but not yet
+   * processed). Stays near 0 on a healthy controller because the pipeline drains quickly and the
+   * queue dedups by event type; climbs when the controller holds leadership but has stopped
+   * processing events ("zombie leader").
+   * <p>
+   * Depth alone is ambiguous: under load a healthy controller also reads &gt; 0 (events queue
+   * behind the in-flight pipeline run), but it keeps draining them, so
+   * {@code ClusterEventStatus...TotalProcessed.EventCounter} advances. A wedged controller instead
+   * shows depth stuck &gt; 0 with that counter flat. The alert threshold and windowing belong in
+   * the alerting layer, not here.
+   * @return The current DEFAULT controller event queue size.
+   */
+  long getControllerEventQueueSizeGauge();
+
+  /**
+   * Reversible 0/1 wedged-controller ("zombie leader") gauge. 1 when the DEFAULT event queue is
+   * non-empty but no pipeline run has completed within the stall threshold, i.e. the controller
+   * holds events but is not processing them; 0 when idle (empty queue) or actively draining.
+   * Unlike the raw queue size, this is producer-rate-independent and does not false-positive on a
+   * busy-but-progressing controller. Gate EKG/alerts on {@code == 1}.
+   * @return 1 if the controller pipeline appears wedged, otherwise 0.
+   */
+  long getControllerPipelineStalledGauge();
+
   // ---- WAGED failure-category counters (mirror of WagedRebalancerMetricCollector) ----
   // Each WAGED HelixRebalanceException increments exactly one of these. The pair
   // {WagedCustomerActionableFailureCounter, WagedInternalFailureCounter} is the recommended

@@ -48,6 +48,15 @@ public void testGetCapacityKeys() {
     Assert.assertEquals(testConfig.getInstanceCapacityKeys(), keys);
   }
 
+  @Test
+  public void testControllerPipelineStallThresholdMsDefaultAndRoundTrip() {
+    ClusterConfig testConfig = new ClusterConfig("testId");
+    // Defaults to 5000ms when unset.
+    Assert.assertEquals(testConfig.getControllerPipelineStallThresholdMs(), 5000L);
+    testConfig.setControllerPipelineStallThresholdMs(30000L);
+    Assert.assertEquals(testConfig.getControllerPipelineStallThresholdMs(), 30000L);
+  }
+
   @Test
   public void testGetCapacityKeysEmpty() {
     ClusterConfig testConfig = new ClusterConfig("testId");

@@ -897,6 +897,108 @@ public void testWagedFallbackInUseGaugeReflectsLatestSetter() {
     Assert.assertEquals(monitor.getWagedFallbackInUseGauge(), 0L);
   }
 
+  @Test
+  public void testControllerEventQueueSizeGaugeStartsAtZero() {
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestControllerEventQueueGaugeCluster");
+    Assert.assertEquals(monitor.getControllerEventQueueSizeGauge(), 0L);
+  }
+
+  @Test
+  public void testControllerEventQueueSizeGaugeReflectsLatestSetter() {
+    ClusterStatusMonitor monitor =
+        new ClusterStatusMonitor("TestControllerEventQueueGaugeSetterCluster");
+    Assert.assertEquals(monitor.getControllerEventQueueSizeGauge(), 0L);
+    monitor.setControllerEventQueueSizeGauge(7L);
+    Assert.assertEquals(monitor.getControllerEventQueueSizeGauge(), 7L);
+    // Gauge is reversible: draining the pipeline takes it back down to 0.
+    monitor.setControllerEventQueueSizeGauge(0L);
+    Assert.assertEquals(monitor.getControllerEventQueueSizeGauge(), 0L);
+  }
+
+  @Test
+  public void testControllerEventQueueSizeGaugeResetsToZeroOnLeadershipChange() {
+    ClusterStatusMonitor monitor =
+        new ClusterStatusMonitor("TestControllerEventQueueGaugeResetCluster");
+    monitor.setControllerEventQueueSizeGauge(5L);
+    Assert.assertEquals(monitor.getControllerEventQueueSizeGauge(), 5L);
+    // reset() runs on leadership change / monitor teardown. Because the monitor instance is reused
+    // across leadership periods, a backlog left over from a prior leader must be zeroed here;
+    // otherwise the re-registered bean would re-report it after re-election until the next
+    // enqueue/dequeue refreshes the gauge.
+    monitor.reset();
+    Assert.assertEquals(monitor.getControllerEventQueueSizeGauge(), 0L);
+  }
+
+  @Test
+  public void testControllerPipelineStalledGaugeStartsAtZero() {
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestPipelineStalledStartCluster");
+    // No queue backlog and no pipeline completion reported yet: no baseline, so not stalled.
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 0L);
+  }
+
+  @Test
+  public void testControllerPipelineStalledGaugeZeroWhenQueueEmpty() {
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestPipelineStalledIdleCluster");
+    // Empty queue but a stale last-completion timestamp: an idle controller is healthy, not wedged.
+    monitor.setControllerEventQueueSizeGauge(0L);
+    monitor.setLastPipelineEndTimestamp(System.currentTimeMillis() - 60000L);
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 0L);
+  }
+
+  @Test
+  public void testControllerPipelineStalledGaugeZeroWhenProgressing() {
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestPipelineStalledBusyCluster");
+    // Non-empty queue but a pipeline completed just now: busy-but-progressing, not wedged.
+    monitor.setControllerEventQueueSizeGauge(5L);
+    monitor.setLastPipelineEndTimestamp(System.currentTimeMillis());
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 0L);
+  }
+
+  @Test
+  public void testControllerPipelineStalledGaugeZeroWithoutBaseline() {
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestPipelineStalledNoBaselineCluster");
+    // Non-empty queue but no pipeline completion ever reported (timestamp 0): treated as no data.
+    monitor.setControllerEventQueueSizeGauge(5L);
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 0L);
+  }
+
+  @Test
+  public void testControllerPipelineStalledGaugeOneWhenWedged() {
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestPipelineStalledWedgedCluster");
+    // Non-empty queue whose pipeline last completed longer ago than the stall threshold (5s):
+    // the controller holds events but is not draining them, i.e. wedged.
+    monitor.setControllerEventQueueSizeGauge(3L);
+    monitor.setLastPipelineEndTimestamp(System.currentTimeMillis() - 6000L);
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 1L);
+  }
+
+  @Test
+  public void testControllerPipelineStalledGaugeResetsToZeroOnLeadershipChange() {
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestPipelineStalledResetCluster");
+    monitor.setControllerEventQueueSizeGauge(3L);
+    monitor.setLastPipelineEndTimestamp(System.currentTimeMillis() - 6000L);
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 1L);
+    // reset() zeroes the progress timestamp so a re-registered bean does not report a stale stall.
+    monitor.reset();
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 0L);
+  }
+
+  @Test
+  public void testControllerPipelineStalledGaugeHonorsConfiguredThreshold() {
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestPipelineStalledThresholdCluster");
+    monitor.setControllerEventQueueSizeGauge(2L);
+    monitor.setLastPipelineEndTimestamp(System.currentTimeMillis() - 6000L); // 6s since last run
+    // With a 10s threshold, a 6s gap is not yet stalled.
+    monitor.setPipelineStallThresholdMs(10000L);
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 0L);
+    // Tightening the threshold to 3s makes the same 6s gap count as stalled.
+    monitor.setPipelineStallThresholdMs(3000L);
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 1L);
+    // A non-positive threshold is ignored (keeps the last valid value), so it stays stalled.
+    monitor.setPipelineStallThresholdMs(0L);
+    Assert.assertEquals(monitor.getControllerPipelineStalledGauge(), 1L);
+  }
+
   @Test
   public void testWagedHardConstraintCountersStartAtZero() {
     ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestWagedHardConstraintCluster");