| package org.apache.helix.integration; |
| |
| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| import java.util.Date; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import org.apache.helix.AccessOption; |
| import org.apache.helix.HelixDataAccessor; |
| import org.apache.helix.PropertyKey; |
| import org.apache.helix.PropertyType; |
| import org.apache.helix.TestHelper; |
| import org.apache.helix.ZkTestHelper; |
| import org.apache.helix.ZkUnitTestBase; |
| import org.apache.helix.integration.manager.ClusterControllerManager; |
| import org.apache.helix.integration.manager.ClusterSpectatorManager; |
| import org.apache.helix.integration.manager.MockParticipantManager; |
| import org.apache.helix.integration.manager.ZkTestManager; |
| import org.apache.helix.manager.zk.CallbackHandler; |
| import org.apache.helix.model.CurrentState; |
| import org.apache.helix.spectator.RoutingTableProvider; |
| import org.apache.helix.tools.ClusterStateVerifier; |
| import org.apache.helix.tools.ClusterVerifiers.BestPossibleExternalViewVerifier; |
| import org.apache.helix.tools.ClusterVerifiers.ZkHelixClusterVerifier; |
| import org.apache.helix.zookeeper.datamodel.ZNRecord; |
| import org.apache.helix.zookeeper.zkclient.IZkChildListener; |
| import org.apache.helix.zookeeper.zkclient.IZkDataListener; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.testng.Assert; |
| import org.testng.annotations.Test; |
| |
| public class TestZkCallbackHandlerLeak extends ZkUnitTestBase { |
| |
| private static Logger LOG = LoggerFactory.getLogger(TestZkCallbackHandlerLeak.class); |
| |
| @Test |
| public void testCbHandlerLeakOnParticipantSessionExpiry() throws Exception { |
| String className = TestHelper.getTestClassName(); |
| String methodName = TestHelper.getTestMethodName(); |
| String clusterName = className + "_" + methodName; |
| final int n = 2; |
| final int r = 2; |
| final int taskResourceCount = 2; |
| |
| System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| |
| TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port |
| "localhost", // participant name prefix |
| "TestDB", // resource name prefix |
| r, // resources |
| 32, // partitions per resource |
| n, // number of nodes |
| 2, // replicas |
| "MasterSlave", true); // do rebalance |
| |
| final ClusterControllerManager controller = |
| new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0"); |
| controller.syncStart(); |
| |
| PropertyKey.Builder keyBuilder = new PropertyKey.Builder(clusterName); |
| // start participants |
| MockParticipantManager[] participants = new MockParticipantManager[n]; |
| for (int i = 0; i < n; i++) { |
| String instanceName = "localhost_" + (12918 + i); |
| |
| participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName); |
| participants[i].syncStart(); |
| |
| // Manually set up task current states |
| for (int j = 0; j < taskResourceCount; j++) { |
| _baseAccessor.create(keyBuilder |
| .taskCurrentState(instanceName, participants[i].getSessionId(), "TestTaskResource_" + j) |
| .toString(), new ZNRecord("TestTaskResource_" + j), AccessOption.PERSISTENT); |
| } |
| } |
| |
| ZkHelixClusterVerifier verifier = new BestPossibleExternalViewVerifier.Builder(clusterName) |
| .setZkClient(_gZkClient) |
| .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME) |
| .build(); |
| Assert.assertTrue(verifier.verifyByPolling()); |
| final MockParticipantManager participantManagerToExpire = participants[1]; |
| |
| // check controller zk-watchers |
| boolean result = TestHelper.verify(() -> { |
| Map<String, Set<String>> watchers = ZkTestHelper.getListenersBySession(ZK_ADDR); |
| Set<String> watchPaths = watchers.get("0x" + controller.getSessionId()); |
| |
| // where n is number of nodes and r is number of resources |
| return watchPaths.size() == (8 + r + (6 + r + taskResourceCount) * n); |
| }, 2000); |
| Assert.assertTrue(result, "Controller has incorrect number of zk-watchers."); |
| |
| // check participant zk-watchers |
| result = TestHelper.verify(() -> { |
| Map<String, Set<String>> watchers = ZkTestHelper.getListenersBySession(ZK_ADDR); |
| Set<String> watchPaths = watchers.get("0x" + participantManagerToExpire.getSessionId()); |
| |
| // participant should have 1 zk-watcher: 1 for MESSAGE |
| return watchPaths.size() == 1; |
| }, 2000); |
| Assert.assertTrue(result, "Participant should have 1 zk-watcher. MESSAGES->HelixTaskExecutor"); |
| |
| // check HelixManager#_handlers |
| int controllerHandlerNb = controller.getHandlers().size(); |
| int particHandlerNb = participantManagerToExpire.getHandlers().size(); |
| Assert.assertEquals(controllerHandlerNb, 8 + 4 * n, |
| "HelixController should have 16 (8+4n) callback handlers for 2 (n) participant"); |
| Assert.assertEquals(particHandlerNb, 1, |
| "HelixParticipant should have 1 (msg->HelixTaskExecutor) callback handlers"); |
| |
| // expire the session of participant |
| System.out.println("Expiring participant session..."); |
| String oldSessionId = participantManagerToExpire.getSessionId(); |
| |
| ZkTestHelper.expireSession(participantManagerToExpire.getZkClient()); |
| String newSessionId = participantManagerToExpire.getSessionId(); |
| System.out.println( |
| "Expired participant session. oldSessionId: " + oldSessionId + ", newSessionId: " |
| + newSessionId); |
| |
| result = |
| ClusterStateVerifier.verifyByPolling(new ClusterStateVerifier.BestPossAndExtViewZkVerifier( |
| ZK_ADDR, clusterName)); |
| Assert.assertTrue(result); |
| |
| // check controller zk-watchers |
| result = TestHelper.verify(() -> { |
| Map<String, Set<String>> watchers = ZkTestHelper.getListenersBySession(ZK_ADDR); |
| Set<String> watchPaths = watchers.get("0x" + controller.getSessionId()); |
| |
| // where n is number of nodes and r is number of resources |
| // one participant is disconnected, and its task current states are removed |
| return watchPaths.size() == (8 + r + (6 + r + taskResourceCount) * (n - 1) + 6 + r); |
| }, 2000); |
| Assert.assertTrue(result, "Controller has incorrect number of zk-watchers after session expiry."); |
| |
| // check participant zk-watchers |
| result = TestHelper.verify(() -> { |
| Map<String, Set<String>> watchers = ZkTestHelper.getListenersBySession(ZK_ADDR); |
| Set<String> watchPaths = watchers.get("0x" + participantManagerToExpire.getSessionId()); |
| |
| // participant should have 1 zk-watcher: 1 for MESSAGE |
| return watchPaths.size() == 1; |
| }, 2000); |
| Assert.assertTrue(result, "Participant should have 1 zk-watcher after session expiry."); |
| |
| // check handlers |
| int handlerNb = controller.getHandlers().size(); |
| Assert.assertEquals(handlerNb, controllerHandlerNb, |
| "controller callback handlers should not increase after participant session expiry"); |
| handlerNb = participantManagerToExpire.getHandlers().size(); |
| Assert.assertEquals(handlerNb, particHandlerNb, |
| "participant callback handlers should not increase after participant session expiry"); |
| |
| // clean up |
| controller.syncStop(); |
| for (int i = 0; i < n; i++) { |
| participants[i].syncStop(); |
| } |
| TestHelper.dropCluster(clusterName, _gZkClient); |
| |
| System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| } |
| |
| @Test |
| public void testCbHandlerLeakOnControllerSessionExpiry() throws Exception { |
| String className = TestHelper.getTestClassName(); |
| String methodName = TestHelper.getTestMethodName(); |
| String clusterName = className + "_" + methodName; |
| final int n = 2; |
| final int r = 1; |
| final int taskResourceCount = 1; |
| |
| System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| |
| TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port |
| "localhost", // participant name prefix |
| "TestDB", // resource name prefix |
| r, // resources |
| 32, // partitions per resource |
| n, // number of nodes |
| 2, // replicas |
| "MasterSlave", true); // do rebalance |
| |
| final ClusterControllerManager controller = |
| new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0"); |
| controller.syncStart(); |
| |
| PropertyKey.Builder keyBuilder = new PropertyKey.Builder(clusterName); |
| // start participants |
| MockParticipantManager[] participants = new MockParticipantManager[n]; |
| for (int i = 0; i < n; i++) { |
| String instanceName = "localhost_" + (12918 + i); |
| |
| participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName); |
| participants[i].syncStart(); |
| // Manually set up task current states |
| for (int j = 0; j < taskResourceCount; j++) { |
| _baseAccessor.create(keyBuilder |
| .taskCurrentState(instanceName, participants[i].getSessionId(), "TestTaskResource_" + j) |
| .toString(), new ZNRecord("TestTaskResource_" + j), AccessOption.PERSISTENT); |
| } |
| } |
| |
| ZkHelixClusterVerifier verifier = |
| new BestPossibleExternalViewVerifier.Builder(clusterName).setZkClient(_gZkClient) |
| .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME) |
| .build(); |
| Assert.assertTrue(verifier.verifyByPolling()); |
| final MockParticipantManager participantManager = participants[0]; |
| |
| // wait until we get all the listeners registered |
| TestHelper.verify(() -> { |
| int controllerHandlerNb = controller.getHandlers().size(); |
| int particHandlerNb = participantManager.getHandlers().size(); |
| if (controllerHandlerNb == 10 && particHandlerNb == 2) |
| return true; |
| else |
| return false; |
| }, 1000); |
| |
| int controllerHandlerNb = controller.getHandlers().size(); |
| int particHandlerNb = participantManager.getHandlers().size(); |
| Assert.assertEquals(controllerHandlerNb, 8 + 4 * n, |
| "HelixController should have 16 (8+4n) callback handlers for 2 participant, but was " |
| + controllerHandlerNb + ", " + printHandlers(controller)); |
| Assert.assertEquals(particHandlerNb, 1, |
| "HelixParticipant should have 1 (msg->HelixTaskExecutor) callback handler, but was " |
| + particHandlerNb + ", " + printHandlers(participantManager)); |
| |
| // expire controller |
| System.out.println("Expiring controller session..."); |
| String oldSessionId = controller.getSessionId(); |
| |
| ZkTestHelper.expireSession(controller.getZkClient()); |
| String newSessionId = controller.getSessionId(); |
| System.out.println( |
| "Expired controller session. oldSessionId: " + oldSessionId + ", newSessionId: " |
| + newSessionId); |
| |
| Assert.assertTrue(verifier.verifyByPolling()); |
| |
| // check controller zk-watchers |
| boolean result = TestHelper.verify(() -> { |
| Map<String, Set<String>> watchers = ZkTestHelper.getListenersBySession(ZK_ADDR); |
| Set<String> watchPaths = watchers.get("0x" + controller.getSessionId()); |
| System.err.println("controller watch paths after session expiry: " + watchPaths.size()); |
| |
| // where r is number of resources and n is number of nodes |
| // task resource count does not attribute to ideal state watch paths |
| int expected = (8 + r + (6 + r + taskResourceCount) * n); |
| return watchPaths.size() == expected; |
| }, 2000); |
| Assert.assertTrue(result, "Controller has incorrect zk-watchers after session expiry."); |
| |
| // check participant zk-watchers |
| result = TestHelper.verify(() -> { |
| Map<String, Set<String>> watchers = ZkTestHelper.getListenersBySession(ZK_ADDR); |
| Set<String> watchPaths = watchers.get("0x" + participantManager.getSessionId()); |
| |
| // participant should have 1 zk-watcher: 1 for MESSAGE |
| return watchPaths.size() == 1; |
| }, 2000); |
| Assert.assertTrue(result, "Participant should have 1 zk-watcher after session expiry."); |
| |
| // check HelixManager#_handlers |
| int handlerNb = controller.getHandlers().size(); |
| Assert.assertEquals(handlerNb, controllerHandlerNb, |
| "controller callback handlers should not increase after participant session expiry, but was " |
| + printHandlers(controller)); |
| handlerNb = participantManager.getHandlers().size(); |
| Assert.assertEquals(handlerNb, particHandlerNb, |
| "participant callback handlers should not increase after participant session expiry, but was " |
| + printHandlers(participantManager)); |
| |
| // clean up |
| controller.syncStop(); |
| for (int i = 0; i < n; i++) { |
| participants[i].syncStop(); |
| } |
| TestHelper.dropCluster(clusterName, _gZkClient); |
| |
| System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| } |
| |
| @Test |
| public void testDanglingCallbackHandler() throws Exception { |
| String className = TestHelper.getTestClassName(); |
| String methodName = TestHelper.getTestMethodName(); |
| String clusterName = className + "_" + methodName; |
| final int n = 3; |
| |
| System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| |
| TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, "localhost", "TestDB", 1, // resource |
| 32, // partitions |
| n, // nodes |
| 2, // replicas |
| "MasterSlave", true); |
| |
| final ClusterControllerManager controller = |
| new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0"); |
| controller.syncStart(); |
| |
| MockParticipantManager[] participants = new MockParticipantManager[n]; |
| for (int i = 0; i < n; i++) { |
| String instanceName = "localhost_" + (12918 + i); |
| participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName); |
| participants[i].syncStart(); |
| } |
| |
| ZkHelixClusterVerifier verifier = |
| new BestPossibleExternalViewVerifier.Builder(clusterName).setZkClient(_gZkClient) |
| .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME) |
| .build(); |
| Assert.assertTrue(verifier.verifyByPolling()); |
| |
| // Routing provider is a spectator in Helix. Currentstate based RP listens on all the |
| // currentstate changes of all the clusters. They are a source of leaking of watch in |
| // Zookeeper server. |
| ClusterSpectatorManager rpManager = new ClusterSpectatorManager(ZK_ADDR, clusterName, "router"); |
| rpManager.syncStart(); |
| RoutingTableProvider rp = new RoutingTableProvider(rpManager, PropertyType.CURRENTSTATES); |
| |
| //TODO: The following three sleep() is not the best practice. On the other hand, we don't have the testing |
| // facilities to avoid them yet. We will enhance later. |
| Thread.sleep(5000); |
| |
| // expire RoutingProvider would create dangling CB |
| LOG.info("expire rp manager session:", rpManager.getSessionId()); |
| ZkTestHelper.expireSession(rpManager.getZkClient()); |
| LOG.info("rp manager new session:", rpManager.getSessionId()); |
| |
| Thread.sleep(5000); |
| |
| MockParticipantManager participantToExpire = participants[0]; |
| String oldSessionId = participantToExpire.getSessionId(); |
| |
| // expire participant session; leaked callback handler used to be not reset() and be removed from ZkClient |
| LOG.info("Expire participant: " + participantToExpire.getInstanceName() + ", session: " |
| + participantToExpire.getSessionId()); |
| ZkTestHelper.expireSession(participantToExpire.getZkClient()); |
| String newSessionId = participantToExpire.getSessionId(); |
| LOG.info(participantToExpire.getInstanceName() + " oldSessionId: " + oldSessionId |
| + ", newSessionId: " + newSessionId); |
| |
| Thread.sleep(5000); |
| Map<String, Set<IZkChildListener>> childListeners = |
| ZkTestHelper.getZkChildListener(rpManager.getZkClient()); |
| for (String path : childListeners.keySet()) { |
| Assert.assertTrue(childListeners.get(path).size() <= 1); |
| } |
| |
| Map<String, List<String>> rpWatchPaths = ZkTestHelper.getZkWatch(rpManager.getZkClient()); |
| List<String> existWatches = rpWatchPaths.get("existWatches"); |
| Assert.assertTrue(existWatches.isEmpty()); |
| |
| // clean up |
| controller.syncStop(); |
| rp.shutdown(); |
| |
| Assert.assertTrue(rpManager.getHandlers().isEmpty(), |
| "HelixManager should not have any callback handlers after shutting down RoutingTableProvider"); |
| |
| rpManager.syncStop(); |
| for (int i = 0; i < n; i++) { |
| participants[i].syncStop(); |
| } |
| TestHelper.dropCluster(clusterName, _gZkClient); |
| |
| System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| } |
| |
| @Test |
| public void testCurrentStatePathLeakingByAsycRemoval() throws Exception { |
| String className = TestHelper.getTestClassName(); |
| String methodName = TestHelper.getTestMethodName(); |
| String clusterName = className + "_" + methodName; |
| final int n = 3; |
| final String zkAddr = ZK_ADDR; |
| final int mJobUpdateCnt = 500; |
| |
| System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| |
| TestHelper.setupCluster(clusterName, zkAddr, 12918, "localhost", "TestDB", 1, // resource |
| 32, // partitions |
| n, // nodes |
| 2, // replicas |
| "MasterSlave", true); |
| |
| final ClusterControllerManager controller = |
| new ClusterControllerManager(zkAddr, clusterName, "controller_0"); |
| controller.syncStart(); |
| |
| MockParticipantManager[] participants = new MockParticipantManager[n]; |
| for (int i = 0; i < n; i++) { |
| String instanceName = "localhost_" + (12918 + i); |
| participants[i] = new MockParticipantManager(zkAddr, clusterName, instanceName); |
| participants[i].syncStart(); |
| } |
| |
| ZkHelixClusterVerifier verifier = |
| new BestPossibleExternalViewVerifier.Builder(clusterName).setZkClient(_gZkClient) |
| .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME) |
| .build(); |
| Assert.assertTrue(verifier.verifyByPolling()); |
| |
| ClusterSpectatorManager rpManager = new ClusterSpectatorManager(ZK_ADDR, clusterName, "router"); |
| rpManager.syncStart(); |
| RoutingTableProvider rp = new RoutingTableProvider(rpManager, PropertyType.CURRENTSTATES); |
| |
| MockParticipantManager jobParticipant = participants[0]; |
| String jobSessionId = jobParticipant.getSessionId(); |
| HelixDataAccessor jobAccessor = jobParticipant.getHelixDataAccessor(); |
| PropertyKey.Builder jobKeyBuilder = new PropertyKey.Builder(clusterName); |
| PropertyKey db0key = |
| jobKeyBuilder.currentState(jobParticipant.getInstanceName(), jobSessionId, "TestDB0"); |
| CurrentState db0 = jobAccessor.getProperty(db0key); |
| PropertyKey jobKey = |
| jobKeyBuilder.currentState(jobParticipant.getInstanceName(), jobSessionId, "BackupQueue"); |
| CurrentState cs = new CurrentState("BackupQueue"); |
| cs.setSessionId(jobSessionId); |
| cs.setStateModelDefRef(db0.getStateModelDefRef()); |
| |
| Map<String, List<String>> rpWatchPaths = ZkTestHelper.getZkWatch(rpManager.getZkClient()); |
| Assert.assertFalse(rpWatchPaths.get("dataWatches").contains(jobKey.getPath())); |
| |
| LOG.info("add job"); |
| for (int i = 0; i < mJobUpdateCnt; i++) { |
| jobAccessor.setProperty(jobKey, cs); |
| } |
| |
| // verify new watcher is installed on the new node |
| boolean result = TestHelper.verify( |
| () -> ZkTestHelper.getListenersByZkPath(ZK_ADDR).keySet().contains(jobKey.getPath()) |
| && ZkTestHelper.getZkWatch(rpManager.getZkClient()).get("dataWatches") |
| .contains(jobKey.getPath()), TestHelper.WAIT_DURATION); |
| Assert.assertTrue(result, |
| "Should get initial clusterConfig callback invoked and add data watchers"); |
| |
| LOG.info("remove job"); |
| jobParticipant.getZkClient().delete(jobKey.getPath()); |
| |
| // validate the job watch is not leaked. |
| Thread.sleep(5000); |
| |
| Map<String, Set<String>> listenersByZkPath = ZkTestHelper.getListenersByZkPath(ZK_ADDR); |
| Assert.assertFalse(listenersByZkPath.keySet().contains(jobKey.getPath())); |
| |
| rpWatchPaths = ZkTestHelper.getZkWatch(rpManager.getZkClient()); |
| List<String> existWatches = rpWatchPaths.get("existWatches"); |
| Assert.assertTrue(existWatches.isEmpty()); |
| |
| // clean up |
| controller.syncStop(); |
| for (int i = 0; i < n; i++) { |
| participants[i].syncStop(); |
| } |
| rp.shutdown(); |
| |
| Assert.assertTrue(rpManager.getHandlers().isEmpty(), |
| "HelixManager should not have any callback handlers after shutting down RoutingTableProvider"); |
| |
| rpManager.syncStop(); |
| TestHelper.dropCluster(clusterName, _gZkClient); |
| |
| System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| } |
| |
| @Test |
| public void testRemoveUserCbHandlerOnPathRemoval() throws Exception { |
| String className = TestHelper.getTestClassName(); |
| String methodName = TestHelper.getTestMethodName(); |
| String clusterName = className + "_" + methodName; |
| final int n = 3; |
| final String zkAddr = ZK_ADDR; |
| System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| |
| TestHelper.setupCluster(clusterName, zkAddr, 12918, "localhost", "TestDB", 1, // resource |
| 32, // partitions |
| n, // nodes |
| 2, // replicas |
| "MasterSlave", true); |
| |
| final ClusterControllerManager controller = |
| new ClusterControllerManager(zkAddr, clusterName, "controller_0"); |
| controller.syncStart(); |
| |
| MockParticipantManager[] participants = new MockParticipantManager[n]; |
| for (int i = 0; i < n; i++) { |
| String instanceName = "localhost_" + (12918 + i); |
| participants[i] = new MockParticipantManager(zkAddr, clusterName, instanceName); |
| participants[i].syncStart(); |
| |
| // register a controller listener on participant_0 |
| if (i == 0) { |
| MockParticipantManager manager = participants[0]; |
| manager.addCurrentStateChangeListener((instanceName1, statesInfo, changeContext) -> { |
| // To change body of implemented methods use File | Settings | File Templates. |
| }, manager.getInstanceName(), manager.getSessionId()); |
| } |
| } |
| |
| ZkHelixClusterVerifier verifier = |
| new BestPossibleExternalViewVerifier.Builder(clusterName).setZkClient(_gZkClient) |
| .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME) |
| .build(); |
| Assert.assertTrue(verifier.verifyByPolling()); |
| |
| MockParticipantManager participantToExpire = participants[0]; |
| String oldSessionId = participantToExpire.getSessionId(); |
| PropertyKey.Builder keyBuilder = new PropertyKey.Builder(clusterName); |
| |
| // check manager#hanlders |
| Assert.assertEquals(participantToExpire.getHandlers().size(), 2, |
| "Should have 2 handlers: CURRENTSTATE/{sessionId}, and MESSAGES"); |
| |
| // check zkclient#listeners |
| Map<String, Set<IZkDataListener>> dataListeners = |
| ZkTestHelper.getZkDataListener(participantToExpire.getZkClient()); |
| Map<String, Set<IZkChildListener>> childListeners = |
| ZkTestHelper.getZkChildListener(participantToExpire.getZkClient()); |
| Assert.assertEquals(dataListeners.size(), 1, |
| "Should have 1 path (CURRENTSTATE/{sessionId}/TestDB0) which has 1 data-listeners"); |
| String path = |
| keyBuilder.currentState(participantToExpire.getInstanceName(), oldSessionId, "TestDB0") |
| .getPath(); |
| Assert.assertEquals(dataListeners.get(path).size(), 1, "Should have 1 data-listeners on path: " |
| + path); |
| Assert |
| .assertEquals(childListeners.size(), 2, |
| "Should have 2 paths (CURRENTSTATE/{sessionId}, and MESSAGES) each of which has 1 child-listener"); |
| path = keyBuilder.currentStates(participantToExpire.getInstanceName(), oldSessionId).getPath(); |
| Assert.assertEquals(childListeners.get(path).size(), 1, |
| "Should have 1 child-listener on path: " + path); |
| path = keyBuilder.messages(participantToExpire.getInstanceName()).getPath(); |
| Assert.assertEquals(childListeners.get(path).size(), 1, |
| "Should have 1 child-listener on path: " + path); |
| path = keyBuilder.controller().getPath(); |
| Assert.assertNull(childListeners.get(path), "Should have no child-listener on path: " + path); |
| |
| // check zookeeper#watches on client side |
| Map<String, List<String>> watchPaths = |
| ZkTestHelper.getZkWatch(participantToExpire.getZkClient()); |
| Assert |
| .assertEquals(watchPaths.get("dataWatches").size(), 3, |
| "Should have 3 data-watches: CURRENTSTATE/{sessionId}, CURRENTSTATE/{sessionId}/TestDB, MESSAGES"); |
| Assert.assertEquals(watchPaths.get("childWatches").size(), 2, |
| "Should have 2 child-watches: MESSAGES, and CURRENTSTATE/{sessionId}"); |
| |
| // expire localhost_12918 |
| System.out.println( |
| "Expire participant: " + participantToExpire.getInstanceName() + ", session: " |
| + participantToExpire.getSessionId()); |
| ZkTestHelper.expireSession(participantToExpire.getZkClient()); |
| String newSessionId = participantToExpire.getSessionId(); |
| System.out.println(participantToExpire.getInstanceName() + " oldSessionId: " + oldSessionId |
| + ", newSessionId: " + newSessionId); |
| verifier = |
| new BestPossibleExternalViewVerifier.Builder(clusterName).setZkClient(_gZkClient) |
| .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME) |
| .build(); |
| Assert.assertTrue(verifier.verifyByPolling()); |
| |
| // check manager#hanlders |
| Assert |
| .assertEquals( |
| participantToExpire.getHandlers().size(), |
| 1, |
| "Should have 1 handlers: MESSAGES. CURRENTSTATE/{sessionId} handler should be removed by CallbackHandler#handleChildChange()"); |
| |
| // check zkclient#listeners |
| dataListeners = ZkTestHelper.getZkDataListener(participantToExpire.getZkClient()); |
| childListeners = ZkTestHelper.getZkChildListener(participantToExpire.getZkClient()); |
| Assert.assertTrue(dataListeners.isEmpty(), "Should have no data-listeners"); |
| Assert |
| .assertEquals( |
| childListeners.size(), |
| 2, |
| "Should have 2 paths (CURRENTSTATE/{oldSessionId}, and MESSAGES). " |
| + "CONTROLLER and MESSAGE has 1 child-listener each. CURRENTSTATE/{oldSessionId} doesn't have listener (ZkClient doesn't remove empty childListener set. probably a ZkClient bug. see ZkClient#unsubscribeChildChange())"); |
| path = keyBuilder.currentStates(participantToExpire.getInstanceName(), oldSessionId).getPath(); |
| Assert.assertEquals(childListeners.get(path).size(), 0, |
| "Should have no child-listener on path: " + path); |
| path = keyBuilder.messages(participantToExpire.getInstanceName()).getPath(); |
| Assert.assertEquals(childListeners.get(path).size(), 1, |
| "Should have 1 child-listener on path: " + path); |
| path = keyBuilder.controller().getPath(); |
| Assert.assertNull(childListeners.get(path), |
| "Should have no child-listener on path: " + path); |
| |
| // check zookeeper#watches on client side |
| watchPaths = ZkTestHelper.getZkWatch(participantToExpire.getZkClient()); |
| Assert.assertEquals(watchPaths.get("dataWatches").size(), 1, |
| "Should have 1 data-watches: MESSAGES"); |
| Assert.assertEquals(watchPaths.get("childWatches").size(), 1, |
| "Should have 1 child-watches: MESSAGES"); |
| |
| // In this test participant0 also register to its own cocurrent state with a callbackhandler |
| // When session expiration happens, the current state parent path would also changes. However, |
| // an exists watch would still be installed by event pushed to ZkCLient event thread by |
| // fireAllEvent() children even path on behalf of old session callbackhandler. By the time this |
| // event gets invoked, the old session callbackhandler was removed, but the event would still |
| // install a exist watch for old session. |
| // The closest similar case in production is that router/controller has an session expiration at |
| // the same time as participant. |
| // Currently there are many places to register watch in Zookeeper over the evolution of Helix |
| // and ZkClient. We plan for further simplify the logic of watch installation next. |
| boolean result = TestHelper.verify(()-> { |
| Map<String, List<String>> wPaths = ZkTestHelper.getZkWatch(participantToExpire.getZkClient()); |
| return wPaths.get("existWatches").size() == 1; |
| }, TestHelper.WAIT_DURATION); |
| Assert.assertTrue(result, |
| "Should have 1 exist-watches: CURRENTSTATE/{oldSessionId}"); |
| |
| // another session expiry on localhost_12918 should clear the two exist-watches on |
| // CURRENTSTATE/{oldSessionId} |
| System.out.println( |
| "Expire participant: " + participantToExpire.getInstanceName() + ", session: " |
| + participantToExpire.getSessionId()); |
| ZkTestHelper.expireSession(participantToExpire.getZkClient()); |
| |
| Assert.assertTrue(verifier.verifyByPolling()); |
| |
| // check zookeeper#watches on client side |
| watchPaths = ZkTestHelper.getZkWatch(participantToExpire.getZkClient()); |
| Assert.assertEquals(watchPaths.get("dataWatches").size(), 1, |
| "Should have 1 data-watches: MESSAGES"); |
| Assert.assertEquals(watchPaths.get("childWatches").size(), 1, |
| "Should have 1 child-watches: MESSAGES"); |
| Assert |
| .assertEquals( |
| watchPaths.get("existWatches").size(), |
| 0, |
| "Should have no exist-watches. exist-watches on CURRENTSTATE/{oldSessionId} and CURRENTSTATE/{oldSessionId}/TestDB0 should be cleared during handleNewSession"); |
| |
| // clean up |
| controller.syncStop(); |
| for (int i = 0; i < n; i++) { |
| participants[i].syncStop(); |
| } |
| TestHelper.dropCluster(clusterName, _gZkClient); |
| |
| System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| } |
| |
| // debug code |
| static String printHandlers(ZkTestManager manager) { |
| StringBuilder sb = new StringBuilder(); |
| List<CallbackHandler> handlers = manager.getHandlers(); |
| sb.append(manager.getInstanceName() + " has " + handlers.size() + " cb-handlers. ["); |
| |
| for (int i = 0; i < handlers.size(); i++) { |
| CallbackHandler handler = handlers.get(i); |
| String path = handler.getPath(); |
| sb.append(path.substring(manager.getClusterName().length() + 1) + ": " |
| + handler.getListener()); |
| if (i < (handlers.size() - 1)) { |
| sb.append(", "); |
| } |
| } |
| sb.append("]"); |
| |
| return sb.toString(); |
| } |
| } |
