This is an automated email from the ASF dual-hosted git repository. jxue pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/helix.git
commit c40dfef76077d0f067493232d68493005a49adf1 Author: LZD-PratyushBhatt <[email protected]> AuthorDate: Sat Jun 28 19:00:57 2025 +0530 Fix global lock contention in DistClusterControllerStateModel caused by Optional.empty() singleton --- .../DistClusterControllerStateModel.java | 7 ++- .../participant/TestDistControllerStateModel.java | 73 ++++++++++++++++++++++ 2 files changed, 78 insertions(+), 2 deletions(-) diff --git a/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java b/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java index 42f0e4b41..8710119d6 100644 --- a/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java +++ b/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java @@ -40,6 +40,9 @@ public class DistClusterControllerStateModel extends AbstractHelixLeaderStandbyS protected Optional<HelixManager> _controllerOpt = Optional.empty(); private final Set<Pipeline.Type> _enabledPipelineTypes; + // dedicated lock object to avoid cross-instance contention from Optional.empty() singleton + private final Object _controllerLock = new Object(); + public DistClusterControllerStateModel(String zkAddr) { this(zkAddr, Sets.newHashSet(Pipeline.Type.DEFAULT, Pipeline.Type.TASK)); } @@ -62,7 +65,7 @@ public class DistClusterControllerStateModel extends AbstractHelixLeaderStandbyS logger.info(controllerName + " becoming leader from standby for " + clusterName); - synchronized (_controllerOpt) { + synchronized (_controllerLock) { if (!_controllerOpt.isPresent()) { HelixManager newController = HelixManagerFactory .getZKHelixManager(clusterName, controllerName, InstanceType.CONTROLLER, _zkAddr); @@ -112,7 +115,7 @@ public class DistClusterControllerStateModel extends AbstractHelixLeaderStandbyS @Override public void reset() { - synchronized (_controllerOpt) { + synchronized (_controllerLock) { if (_controllerOpt.isPresent()) { logger.info("Disconnecting controller: " + _controllerOpt.get().getInstanceName() + " for " + _controllerOpt.get().getClusterName()); diff --git a/helix-core/src/test/java/org/apache/helix/participant/TestDistControllerStateModel.java b/helix-core/src/test/java/org/apache/helix/participant/TestDistControllerStateModel.java index d07aab6b0..3e61bd854 100644 --- a/helix-core/src/test/java/org/apache/helix/participant/TestDistControllerStateModel.java +++ b/helix-core/src/test/java/org/apache/helix/participant/TestDistControllerStateModel.java @@ -27,10 +27,18 @@ import org.apache.helix.model.Message; import org.apache.helix.model.Message.MessageType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; +import java.lang.reflect.Field; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + public class TestDistControllerStateModel extends ZkUnitTestBase { private static Logger LOG = LoggerFactory.getLogger(TestDistControllerStateModel.class); @@ -124,4 +132,69 @@ public class TestDistControllerStateModel extends ZkUnitTestBase { stateModel.reset(); } + /** + * Test to verify that different DistClusterControllerStateModel instances + * use separate lock objects, ensuring no cross-instance blocking. + */ + @Test() + public void testNoSharedLockAcrossInstances() throws Exception { + LOG.info("Testing that lock objects are not shared across DistClusterControllerStateModel instances"); + + // Verify different instances have different lock objects + DistClusterControllerStateModel instance1 = new DistClusterControllerStateModel(ZK_ADDR); + DistClusterControllerStateModel instance2 = new DistClusterControllerStateModel(ZK_ADDR); + + Field lockField = DistClusterControllerStateModel.class.getDeclaredField("_controllerLock"); + lockField.setAccessible(true); + + Object lock1 = lockField.get(instance1); + Object lock2 = lockField.get(instance2); + + Assert.assertNotNull(lock1, "First instance should have a lock object"); + Assert.assertNotNull(lock2, "Second instance should have a lock object"); + Assert.assertNotSame(lock1, lock2, "Different instances must have different lock objects"); + + // Verify concurrent access doesn't block across instances + final int NUM_INSTANCES = 10; + ExecutorService executor = Executors.newFixedThreadPool(NUM_INSTANCES); + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch completionLatch = new CountDownLatch(NUM_INSTANCES); + AtomicInteger completedInstances = new AtomicInteger(0); + + for (int i = 0; i < NUM_INSTANCES; i++) { + final int instanceId = i; + final DistClusterControllerStateModel instance = new DistClusterControllerStateModel(ZK_ADDR); + + executor.submit(() -> { + try { + startLatch.await(); // wait for all threads to be ready + + // Simulate state transition operations that would use the lock + synchronized (lockField.get(instance)) { + // hold the lock here briefly to simulate real state transition work + Thread.sleep(100); + completedInstances.incrementAndGet(); + } + + } catch (Exception e) { + LOG.error("Instance {} failed during concurrent test", instanceId, e); + } finally { + completionLatch.countDown(); + } + }); + } + + // start all threads simultaneously + startLatch.countDown(); + + // All instances should complete within reasonable time since they don't block each other + boolean allCompleted = completionLatch.await(2, TimeUnit.SECONDS); + + executor.shutdown(); + executor.awaitTermination(2, TimeUnit.SECONDS); + + Assert.assertTrue(allCompleted, "All instances should complete without blocking each other"); + Assert.assertEquals(completedInstances.get(), NUM_INSTANCES, + "All instances should successfully complete their synchronized work"); + } }
