Add ControllerEventQueueSizeGauge to detect a wedged ("zombie leader" controller)

helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java

@@ -1342,6 +1342,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
@@ -1579,6 +1594,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);

helix-core/src/main/java/org/apache/helix/monitoring/mbeans/ClusterStatusMonitor.java

@@ -94,6 +94,10 @@ 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);
 
   // 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.
@@ -1467,6 +1471,16 @@ 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);
+  }
+
   @Override
   public long getRebalanceFailureCounter() {
     return _rebalanceFailureCount.get();
@@ -1482,6 +1496,11 @@ public long getContinuousTaskRebalanceFailureCount() {
     return _continuousTaskRebalanceFailureCount.get();
   }
 
+  @Override
+  public long getControllerEventQueueSizeGauge() {
+    return _controllerEventQueueSizeGauge.get();
+  }
+
   @Override
   public long getWagedCustomerActionableFailureCounter() {
     return _wagedCustomerActionableFailureCount.get();

helix-core/src/main/java/org/apache/helix/monitoring/mbeans/ClusterStatusMonitorMBean.java

@@ -97,6 +97,15 @@ 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"). Alert on a sustained average above ~0.
+   * @return The current DEFAULT controller event queue size.
+   */
+  long getControllerEventQueueSizeGauge();
+
   // ---- WAGED failure-category counters (mirror of WagedRebalancerMetricCollector) ----
   // Each WAGED HelixRebalanceException increments exactly one of these. The pair
   // {WagedCustomerActionableFailureCounter, WagedInternalFailureCounter} is the recommended

helix-core/src/test/java/org/apache/helix/monitoring/mbeans/TestClusterStatusMonitor.java

@@ -897,6 +897,24 @@ 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 testWagedHardConstraintCountersStartAtZero() {
     ClusterStatusMonitor monitor = new ClusterStatusMonitor("TestWagedHardConstraintCluster");