Adding the sensors in DistClusterControllerStateModel

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

@@ -87,6 +87,8 @@ public class ClusterStatusMonitor implements ClusterStatusMonitorMBean {
   private AtomicLong _rebalanceFailureCount = new AtomicLong(0L);
   private AtomicLong _continuousResourceRebalanceFailureCount = new AtomicLong(0L);
   private AtomicLong _continuousTaskRebalanceFailureCount = new AtomicLong(0L);
+  private AtomicLong _leadershipFailureCount = new AtomicLong(0L);
+  private AtomicLong _stillLeaderDuringResetCount = new AtomicLong(0L);
 
   private final ConcurrentHashMap<String, ResourceMonitor> _resourceMonitorMap =
       new ConcurrentHashMap<>();
@@ -660,6 +662,8 @@ public void reset() {
       _rebalanceFailureCount.set(0L);
       _continuousResourceRebalanceFailureCount.set(0L);
       _continuousTaskRebalanceFailureCount.set(0L);
+      _leadershipFailureCount.set(0L);
+      _stillLeaderDuringResetCount.set(0L);
     } catch (Exception e) {
       LOG.error("Fail to reset ClusterStatusMonitor, cluster: " + _clusterName, e);
     }
@@ -1137,4 +1141,28 @@ public long getNumOfResourcesRebalanceThrottledGauge() {
     }
     return total;
   }
+
+  @Override
+  public long getLeadershipFailureCounter() {
+    return _leadershipFailureCount.get();
+  }
+
+  @Override
+  public long getStillLeaderDuringResetCounter() {
+    return _stillLeaderDuringResetCount.get();
+  }
+
+  /**
+   * Report a leadership failure for distributed controller
+   */
+  public void reportLeadershipFailure() {
+    _leadershipFailureCount.incrementAndGet();
+  }
+
+  /**
+   * Report when controller is still leader during reset
+   */
+  public void reportStillLeaderDuringReset() {
+    _stillLeaderDuringResetCount.incrementAndGet();
+  }
 }

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

@@ -147,4 +147,14 @@ public interface ClusterStatusMonitorMBean extends SensorNameProvider {
    * state partition is larger than configured threshold (default is 1).
    */
   long getNumOfResourcesRebalanceThrottledGauge();
+
+  /**
+   * @return The number of leadership failures for distributed controllers
+   */
+  long getLeadershipFailureCounter();
+
+  /**
+   * @return The number of times controller was still leader during reset
+   */
+  long getStillLeaderDuringResetCounter();
 }

helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java

@@ -29,6 +29,7 @@
 import org.apache.helix.NotificationContext;
 import org.apache.helix.controller.pipeline.Pipeline;
 import org.apache.helix.model.Message;
+import org.apache.helix.monitoring.mbeans.ClusterStatusMonitor;
 import org.apache.helix.participant.statemachine.StateModelInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -37,8 +38,14 @@
 @StateModelInfo(initialState = "OFFLINE", states = {"LEADER", "STANDBY"})
 public class DistClusterControllerStateModel extends AbstractHelixLeaderStandbyStateModel {
   private static Logger logger = LoggerFactory.getLogger(DistClusterControllerStateModel.class);
+
+  // Constants for metric types
+  private static final String METRIC_LEADERSHIP_FAILURE = "leadership_failure";
+  private static final String METRIC_STILL_LEADER_DURING_RESET = "still_leader_during_reset";
+
   protected volatile Optional<HelixManager> _controllerOpt = Optional.empty();
   private final Set<Pipeline.Type> _enabledPipelineTypes;
+  private ClusterStatusMonitor _clusterStatusMonitor;
 
   // dedicated lock object to avoid cross-instance contention from Optional.empty() singleton
   private final Object _controllerLock = new Object();
@@ -73,6 +80,13 @@ public void onBecomeLeaderFromStandby(Message message, NotificationContext conte
         newController.connect();
         newController.startTimerTasks();
         _controllerOpt = Optional.of(newController);
+        if (!newController.isLeader()) {
+          logger.warn("Controller Leader session is not the same as the current session for {}. "
+              + "This should not happen. Controller: {}", clusterName ,controllerName);
+
+          // Publish metrics through ClusterStatusMonitor when not a leader
+          publishControllerMetric(clusterName, METRIC_LEADERSHIP_FAILURE);
+        }
         logStateTransition("STANDBY", "LEADER", clusterName, controllerName);
       } else {
         logger.error("controller already exists:" + _controllerOpt.get().getInstanceName() + " for "
@@ -117,11 +131,53 @@ public String getStateModeInstanceDescription(String partitionName, String insta
   public void reset() {
     synchronized (_controllerLock) {
       if (_controllerOpt.isPresent()) {
+        String clusterName = _controllerOpt.get().getClusterName();
         logger.info("Disconnecting controller: " + _controllerOpt.get().getInstanceName() + " for "
             + _controllerOpt.get().getClusterName());
         _controllerOpt.get().disconnect();
+        if(_controllerOpt.get().isLeader()) {
+          logger.warn("Controller is still leader after disconnecting: {} for {}",
+              _controllerOpt.get().getInstanceName(), _controllerOpt.get().getClusterName());
+
+          // Publish metrics when controller is still leader during reset
+          publishControllerMetric(clusterName, METRIC_STILL_LEADER_DURING_RESET);
+        }
         _controllerOpt = Optional.empty();
       }
+
+      // Clean up cluster status monitor
+      if (_clusterStatusMonitor != null) {
+        _clusterStatusMonitor.reset();
+        _clusterStatusMonitor = null;
+      }
+    }
+  }
+
+  /**
+   * Publishes controller metrics through the managed ClusterStatusMonitor
+   * @param clusterName the cluster name
+   * @param metricType the type of metric to publish
+   */
+  private void publishControllerMetric(String clusterName, String metricType) {
+    try {
+      // Ensure we have a monitor for this cluster
+      if (_clusterStatusMonitor == null) {
+          _clusterStatusMonitor = new ClusterStatusMonitor(clusterName);
+          _clusterStatusMonitor.active();
+      }
+
+      switch (metricType) {
+        case METRIC_LEADERSHIP_FAILURE:
+          _clusterStatusMonitor.reportLeadershipFailure();
+          break;
+        case METRIC_STILL_LEADER_DURING_RESET:
+          _clusterStatusMonitor.reportStillLeaderDuringReset();
+          break;
+        default:
+          logger.warn("Unknown metric type: {} for cluster {}", metricType, clusterName);
+      }
+    } catch (Exception e) {
+      logger.error("Failed to publish {} metric for cluster {}", metricType, clusterName, e);
     }
   }
 }

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

@@ -610,4 +610,75 @@ private void verifyMessageMetrics(ClusterStatusMonitor monitor, Map<String, Doub
       }
     }
   }
+
+  @Test
+  public void testLeadershipFailureMetrics() throws Exception {
+    String className = TestHelper.getTestClassName();
+    String methodName = TestHelper.getTestMethodName();
+    String clusterName = className + "_" + methodName;
+
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor(clusterName);
+    monitor.active();
+    ObjectName clusterMonitorObjName = monitor.getObjectName(monitor.clusterBeanName());
+
+    Assert.assertTrue(_server.isRegistered(clusterMonitorObjName));
+
+    // Initial state - leadership failure counter should be 0
+    Object initialCount = _server.getAttribute(clusterMonitorObjName, "LeadershipFailureCounter");
+    Assert.assertTrue(initialCount instanceof Long);
+    Assert.assertEquals(initialCount, 0L);
+
+    // Report leadership failure multiple times
+    monitor.reportLeadershipFailure();
+    monitor.reportLeadershipFailure();
+    monitor.reportLeadershipFailure();
+
+    // Verify counter increased
+    Object updatedCount = _server.getAttribute(clusterMonitorObjName, "LeadershipFailureCounter");
+    Assert.assertTrue(updatedCount instanceof Long);
+    Assert.assertEquals(updatedCount, 3L);
+
+    // Verify getter method returns same value
+    Assert.assertEquals(monitor.getLeadershipFailureCounter(), 3L);
+
+    // Clean up
+    monitor.reset();
+    Assert.assertFalse(_server.isRegistered(clusterMonitorObjName),
+        "Failed to unregister ClusterStatusMonitor.");
+  }
+
+  @Test
+  public void testStillLeaderDuringResetMetrics() throws Exception {
+    String className = TestHelper.getTestClassName();
+    String methodName = TestHelper.getTestMethodName();
+    String clusterName = className + "_" + methodName;
+
+    ClusterStatusMonitor monitor = new ClusterStatusMonitor(clusterName);
+    monitor.active();
+    ObjectName clusterMonitorObjName = monitor.getObjectName(monitor.clusterBeanName());
+
+    Assert.assertTrue(_server.isRegistered(clusterMonitorObjName));
+
+    // Initial state - still leader during reset counter should be 0
+    Object initialCount = _server.getAttribute(clusterMonitorObjName, "StillLeaderDuringResetCounter");
+    Assert.assertTrue(initialCount instanceof Long);
+    Assert.assertEquals((Long) initialCount, Long.valueOf(0));
+
+    // Report still leader during reset multiple times
+    monitor.reportStillLeaderDuringReset();
+    monitor.reportStillLeaderDuringReset();
+
+    // Verify counter increased
+    Object updatedCount = _server.getAttribute(clusterMonitorObjName, "StillLeaderDuringResetCounter");
+    Assert.assertTrue(updatedCount instanceof Long);
+    Assert.assertEquals((Long) updatedCount, Long.valueOf(2));
+
+    // Verify getter method returns same value
+    Assert.assertEquals(monitor.getStillLeaderDuringResetCounter(), 2L);
+
+    // Clean up
+    monitor.reset();
+    Assert.assertFalse(_server.isRegistered(clusterMonitorObjName),
+        "Failed to unregister ClusterStatusMonitor.");
+  }
 }

helix-core/src/test/java/org/apache/helix/participant/TestDistControllerStateModel.java

@@ -21,6 +21,7 @@
 
 import org.apache.helix.NotificationContext;
 import org.apache.helix.TestHelper;
+import org.apache.helix.monitoring.mbeans.ClusterStatusMonitor;
 import org.apache.helix.zookeeper.datamodel.ZNRecord;
 import org.apache.helix.ZkUnitTestBase;
 import org.apache.helix.model.Message;
@@ -265,4 +266,36 @@ public void testExplicitLockIndependence() throws Exception {
     Assert.assertTrue(instance1Interrupted.get(),
         "Instance1 should have been interrupted while holding its lock");
   }
+
+  @Test()
+  public void testOnBecomeLeaderFromStandby_whenMultipleInstancesTrigger() throws Exception {
+    // First controller becomes leader
+    Message message = new Message(MessageType.STATE_TRANSITION, "0");
+    message.setPartitionName(clusterName);
+    message.setTgtName("controller_0");
+    stateModel.onBecomeLeaderFromStandby(message, new NotificationContext(null));
+
+    // Second controller attempts to become leader
+    message = new Message(MessageType.STATE_TRANSITION, "1");
+    message.setPartitionName(clusterName);
+    message.setTgtName("controller_1");
+    DistClusterControllerStateModel stateModel2 = new DistClusterControllerStateModel(ZK_ADDR);
+    stateModel2.onBecomeLeaderFromStandby(message, new NotificationContext(null));
+
+    // Verify leadership states
+    // controller_0 is the leader of clusterName.
+    Assert.assertTrue(stateModel._controllerOpt.get().isLeader());
+    // controller_1 was not able to become leader because controller_0 was already the leader.
+    Assert.assertFalse(stateModel2._controllerOpt.get().isLeader());
+
+    // Verify that leadership failure metric was reported by accessing the internal monitor
+    // Use reflection to get the internal ClusterStatusMonitor from stateModel2
+    Field monitorField = DistClusterControllerStateModel.class.getDeclaredField("_clusterStatusMonitor");
+    monitorField.setAccessible(true);
+    ClusterStatusMonitor internalMonitor = (ClusterStatusMonitor) monitorField.get(stateModel2);
+
+    // The monitor should have been created and the metric should be incremented
+    Assert.assertEquals(internalMonitor.getLeadershipFailureCounter(), 1,
+        "Leadership failure metric should be incremented when second controller fails to become leader");
+  }
 }