diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java index 50a9f04390..988ca5adc0 100644 --- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java +++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java @@ -36,6 +36,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Queue; import java.util.Set; import java.util.TreeMap; import java.util.UUID; @@ -128,6 +129,10 @@ public class ZKHelixAdmin implements HelixAdmin { public static final String CONNECTION_TIMEOUT = "helixAdmin.timeOutInSec"; private static final String MAINTENANCE_ZNODE_ID = "maintenance"; private static final int DEFAULT_SUPERCLUSTER_REPLICA = 3; + // Batch size for dropInstance subtree deletion. Sized so each multi() packet + // stays well under jute.maxbuffer (4 MB default): ~240 bytes/op * 1000 ops + // ~= 240 KB. See dropInstancePathsRecursively for context. + private static final int DROP_INSTANCE_DELETE_BATCH_SIZE = 1000; private static final ImmutableSet INSTANCE_OPERATION_TO_EXCLUDE_FROM_ASSIGNMENT = ImmutableSet.of(InstanceConstants.InstanceOperation.EVACUATE, @@ -266,15 +271,15 @@ public void dropInstance(String clusterName, InstanceConfig instanceConfig) { String instanceName = instanceConfig.getInstanceName(); String instanceConfigPath = PropertyPathBuilder.instanceConfig(clusterName, instanceName); - if (!_zkClient.exists(instanceConfigPath)) { - throw new HelixException( - "Node " + instanceName + " does not exist in config for cluster " + clusterName); - } - String instancePath = PropertyPathBuilder.instance(clusterName, instanceName); - if (!_zkClient.exists(instancePath)) { + boolean hasConfig = _zkClient.exists(instanceConfigPath); + boolean hasInstance = _zkClient.exists(instancePath); + // dropInstancePathsRecursively is no longer atomic (config is deleted before + // the subtree). A retry after a partial drop may find the config already + // gone but the subtree still present; treat that as a resume case. + if (!hasConfig && !hasInstance) { throw new HelixException( - "Node " + instanceName + " does not exist in instances for cluster " + clusterName); + "Node " + instanceName + " does not exist in config for cluster " + clusterName); } String liveInstancePath = PropertyPathBuilder.liveInstance(clusterName, instanceName); @@ -286,13 +291,37 @@ public void dropInstance(String clusterName, InstanceConfig instanceConfig) { dropInstancePathsRecursively(clusterName, instanceName); } + // Two-phase drop to avoid jute.maxbuffer violations on instances that have + // accumulated large subtrees (e.g. tens of thousands of MESSAGES, CURRENTSTATES, + // TASKCURRENTSTATES). The previous single deleteRecursivelyAtomic([instance, config]) + // built one multi() packet whose size grew O(#znodes); on instances with ~13K + // messages it crossed the 4 MB jute.maxbuffer limit, which ZK surfaces as + // CONNECTIONLOSS. The default 24h ZK retry timeout then pinned Jetty threads + // until the REST pool was exhausted. + // + // Phase 1: delete InstanceConfig first. This makes the instance non-Assignable, + // so the controller stops generating new state-transition messages while the + // subtree delete is in flight. + // Phase 2: delete the /INSTANCES/{instance} subtree in batched multi() calls + // sized below jute.maxbuffer. + // + // Trade-off: this is no longer atomic. If the JVM dies between Phase 1 and + // the end of Phase 2, a stale /INSTANCES/{instance} subtree remains. The next + // dropInstance call (or a follow-up retry) is idempotent and finishes the + // cleanup. dropInstance's own existence check is relaxed below to allow the + // resume case where InstanceConfig is already gone. private void dropInstancePathsRecursively(String clusterName, String instanceName) { String instanceConfigPath = PropertyPathBuilder.instanceConfig(clusterName, instanceName); String instancePath = PropertyPathBuilder.instance(clusterName, instanceName); int retryCnt = 0; while (true) { try { - _zkClient.deleteRecursivelyAtomic(Arrays.asList(instancePath, instanceConfigPath)); + // Phase 1 + if (_zkClient.exists(instanceConfigPath)) { + _zkClient.delete(instanceConfigPath); + } + // Phase 2 + deleteInstanceSubtreeBatched(instancePath); return; } catch (ZkClientException e) { if (retryCnt < 3 && e.getCause() instanceof ZkException && e.getCause() @@ -313,6 +342,97 @@ private void dropInstancePathsRecursively(String clusterName, String instanceNam } } + // Delete the given path and all descendants using batched multi() calls. Each + // batch is sized so the on-wire packet stays comfortably below jute.maxbuffer. + // NoNode errors inside a batch are tolerated to keep retries idempotent. + private void deleteInstanceSubtreeBatched(String rootPath) { + if (!_zkClient.exists(rootPath)) { + return; + } + List orderedPaths = collectSubtreeChildrenFirst(rootPath); + if (orderedPaths.isEmpty()) { + return; + } + for (int i = 0; i < orderedPaths.size(); i += DROP_INSTANCE_DELETE_BATCH_SIZE) { + int end = Math.min(i + DROP_INSTANCE_DELETE_BATCH_SIZE, orderedPaths.size()); + List ops = new ArrayList<>(end - i); + for (int j = i; j < end; j++) { + ops.add(Op.delete(orderedPaths.get(j), -1)); + } + List opResults; + try { + opResults = _zkClient.multi(ops); + } catch (Exception e) { + throw new ZkClientException( + "Failed batched delete for subtree " + rootPath + " (batch starting at index " + i + + ", size " + ops.size() + ")", e); + } + Map failedPathsMap = new HashMap<>(); + Map notEmptyPathsMap = new HashMap<>(); + for (int k = 0; k < opResults.size(); k++) { + if (opResults.get(k) instanceof OpResult.ErrorResult) { + KeeperException.Code code = KeeperException.Code + .get(((OpResult.ErrorResult) opResults.get(k)).getErr()); + if (code == KeeperException.Code.OK || code == KeeperException.Code.NONODE) { + // NoNode is tolerated: a previous partial delete or concurrent + // delete may have already removed this znode. + continue; + } + if (code == KeeperException.Code.NOTEMPTY) { + // NotEmpty surfaces when a child znode (typically ParticipantHistory + // written by the controller) appears under a parent we are deleting. + // Bubble this up in the same exception shape the legacy + // deleteRecursivelyAtomic produced so the existing retry loop in + // dropInstancePathsRecursively re-walks and retries. + notEmptyPathsMap.put(ops.get(k).getPath(), code); + } else { + failedPathsMap.put(ops.get(k).getPath(), code); + } + } + } + if (!notEmptyPathsMap.isEmpty()) { + String firstNotEmptyPath = notEmptyPathsMap.keySet().iterator().next(); + throw new ZkClientException( + "Batched delete for subtree " + rootPath + " hit NotEmpty on " + notEmptyPathsMap + .keySet(), + new ZkException( + "Batched delete for subtree " + rootPath + " hit NotEmpty on " + notEmptyPathsMap + .keySet(), + new KeeperException.NotEmptyException(firstNotEmptyPath))); + } + if (!failedPathsMap.isEmpty()) { + throw new ZkClientException( + "Batched delete for subtree " + rootPath + " failed with errors: " + failedPathsMap); + } + } + } + + // BFS-walk the subtree rooted at path and return all znode paths ordered so + // children come before their parents (safe for sequential delete). NoNode + // during traversal is tolerated; the missing branch is skipped. + private List collectSubtreeChildrenFirst(String path) { + List orderedPaths = new ArrayList<>(); + Queue queue = new LinkedList<>(); + queue.offer(path); + while (!queue.isEmpty()) { + String node = queue.poll(); + List children; + try { + children = _zkClient.getChildren(node); + } catch (ZkNoNodeException e) { + continue; + } + if (children != null) { + for (String child : children) { + queue.offer(node + "/" + child); + } + } + orderedPaths.add(node); + } + Collections.reverse(orderedPaths); + return orderedPaths; + } + /** * Please note that the purge function should only be called when there is no new instance * joining happening in the cluster. The reason is that current implementation is not thread safe, diff --git a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java index b60b909a26..03e9799cb6 100644 --- a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java +++ b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java @@ -224,20 +224,21 @@ public void testZkHelixAdmin() { } // Tests that ZkClientException thrown from ZkClient should be caught - // and it should be converted HelixException to be rethrown + // and it should be converted HelixException to be rethrown. + // dropInstance now does a two-phase batched delete (config first, then + // subtree via multi() batches). Simulate the racy NotEmpty case by having + // multi() return an OpResult.ErrorResult with NOTEMPTY for the parent znode. String instancePath = PropertyPathBuilder.instance(clusterName, config.getInstanceName()); String instanceConfigPath = PropertyPathBuilder.instanceConfig(clusterName, instanceName); String liveInstancePath = PropertyPathBuilder.liveInstance(clusterName, instanceName); RealmAwareZkClient mockZkClient = Mockito.mock(RealmAwareZkClient.class); - // Mock the exists() method to let dropInstance() reach deleteRecursively(). Mockito.when(mockZkClient.exists(instanceConfigPath)).thenReturn(true); Mockito.when(mockZkClient.exists(instancePath)).thenReturn(true); Mockito.when(mockZkClient.exists(liveInstancePath)).thenReturn(false); - Mockito.doThrow(new ZkClientException("ZkClientException: failed to delete " + instancePath, - new ZkException("ZkException: failed to delete " + instancePath, - new KeeperException.NotEmptyException( - "NotEmptyException: directory" + instancePath + " is not empty")))) - .when(mockZkClient).deleteRecursivelyAtomic(Arrays.asList(instancePath, instanceConfigPath)); + Mockito.when(mockZkClient.getChildren(instancePath)).thenReturn(Collections.emptyList()); + Mockito.when(mockZkClient.multi(Mockito.anyIterable())).thenReturn(Collections.singletonList( + (org.apache.zookeeper.OpResult) new org.apache.zookeeper.OpResult.ErrorResult( + KeeperException.Code.NOTEMPTY.intValue()))); HelixAdmin helixAdminMock = new ZKHelixAdmin(mockZkClient); try { @@ -1367,4 +1368,208 @@ public void testDropInstance() { System.out.println("End test :" + TestHelper.getTestMethodName()); } + + // Verifies dropInstance handles a subtree larger than DROP_INSTANCE_DELETE_BATCH_SIZE + // (1000 ops) by using batched multi() calls. Reproduces the production scenario + // where an instance accumulates large numbers of MESSAGES; the legacy single + // deleteRecursivelyAtomic() built one multi() packet that crossed jute.maxbuffer. + @Test + public void testDropInstanceWithLargeMessageSubtree() { + System.out.println("Start test :" + TestHelper.getTestMethodName()); + final String clusterName = "TestDropInstanceLargeSubtree"; + final String instanceName = "host_with_many_messages"; + final int numMessages = 2500; // > 2 batches of 1000 + + HelixAdmin admin = new ZKHelixAdmin(_gZkClient); + admin.addCluster(clusterName, true); + admin.addInstance(clusterName, new InstanceConfig(instanceName)); + + // Pre-populate /INSTANCES/{instance}/MESSAGES with many znodes + String messagesPath = PropertyPathBuilder.instanceMessage(clusterName, instanceName); + for (int i = 0; i < numMessages; i++) { + _gZkClient.createPersistent(messagesPath + "/msg-" + i); + } + AssertJUnit.assertEquals(numMessages, _gZkClient.getChildren(messagesPath).size()); + + admin.dropInstance(clusterName, new InstanceConfig(instanceName)); + + String instancePath = PropertyPathBuilder.instance(clusterName, instanceName); + String instanceConfigPath = PropertyPathBuilder.instanceConfig(clusterName, instanceName); + AssertJUnit.assertFalse("instance subtree should be gone", _gZkClient.exists(instancePath)); + AssertJUnit.assertFalse("instance config should be gone", _gZkClient.exists(instanceConfigPath)); + AssertJUnit + .assertTrue("cluster instance list should be empty", admin.getInstancesInCluster(clusterName).isEmpty()); + + _gSetupTool.deleteCluster(clusterName); + System.out.println("End test :" + TestHelper.getTestMethodName()); + } + + // Resume case: if a previous dropInstance partially completed (config deleted + // but subtree delete failed), a follow-up dropInstance should clean up the + // remaining subtree instead of erroring on "config does not exist". + @Test + public void testDropInstanceResumesAfterPartialDelete() { + System.out.println("Start test :" + TestHelper.getTestMethodName()); + final String clusterName = "TestDropInstanceResume"; + final String instanceName = "host_partial"; + + HelixAdmin admin = new ZKHelixAdmin(_gZkClient); + admin.addCluster(clusterName, true); + admin.addInstance(clusterName, new InstanceConfig(instanceName)); + String messagesPath = PropertyPathBuilder.instanceMessage(clusterName, instanceName); + _gZkClient.createPersistent(messagesPath + "/leftover-msg"); + + // Simulate a prior partial drop: InstanceConfig already deleted, subtree remains. + String instanceConfigPath = PropertyPathBuilder.instanceConfig(clusterName, instanceName); + _gZkClient.delete(instanceConfigPath); + AssertJUnit.assertFalse(_gZkClient.exists(instanceConfigPath)); + AssertJUnit.assertTrue(_gZkClient.exists(PropertyPathBuilder.instance(clusterName, instanceName))); + + // Resume should succeed and clean up the leftover subtree. + admin.dropInstance(clusterName, new InstanceConfig(instanceName)); + + AssertJUnit.assertFalse(_gZkClient.exists(PropertyPathBuilder.instance(clusterName, instanceName))); + _gSetupTool.deleteCluster(clusterName); + System.out.println("End test :" + TestHelper.getTestMethodName()); + } + + // Realistic instance shape: addInstance creates 7 standard subdirs (MESSAGES, + // CURRENTSTATES, TASKCURRENTSTATES, CUSTOMIZEDSTATES, ERRORS, STATUSUPDATES, + // HISTORY) plus ParticipantHistory. Populate nested children at depth>=2 under + // CURRENTSTATES (sessionId/resource) to verify children-first BFS ordering + // works for non-trivial trees. + @Test + public void testDropInstanceWithDeepSubtreeShape() { + System.out.println("Start test :" + TestHelper.getTestMethodName()); + final String clusterName = "TestDropInstanceDeepShape"; + final String instanceName = "host_deep"; + + HelixAdmin admin = new ZKHelixAdmin(_gZkClient); + admin.addCluster(clusterName, true); + admin.addInstance(clusterName, new InstanceConfig(instanceName)); + + // depth>=2 znodes under CURRENTSTATES: /CURRENTSTATES/{sessionId}/{resource} + String csPath = PropertyPathBuilder.instanceCurrentState(clusterName, instanceName); + String session = "session-1"; + _gZkClient.createPersistent(csPath + "/" + session); + for (int i = 0; i < 50; i++) { + _gZkClient.createPersistent(csPath + "/" + session + "/resource-" + i); + } + // Mixed leaf znodes under MESSAGES and ERRORS + String msgPath = PropertyPathBuilder.instanceMessage(clusterName, instanceName); + for (int i = 0; i < 100; i++) { + _gZkClient.createPersistent(msgPath + "/msg-" + i); + } + String errPath = PropertyPathBuilder.instanceError(clusterName, instanceName); + _gZkClient.createPersistent(errPath + "/" + session); + _gZkClient.createPersistent(errPath + "/" + session + "/res-1"); + + admin.dropInstance(clusterName, new InstanceConfig(instanceName)); + + AssertJUnit.assertFalse(_gZkClient.exists(PropertyPathBuilder.instance(clusterName, instanceName))); + AssertJUnit.assertFalse(_gZkClient.exists(PropertyPathBuilder.instanceConfig(clusterName, instanceName))); + _gSetupTool.deleteCluster(clusterName); + System.out.println("End test :" + TestHelper.getTestMethodName()); + } + + // Boundary: small subtree fits in a single multi() batch. Verifies the + // single-batch path (loop runs once) is exercised end-to-end. + @Test + public void testDropInstanceFitsInSingleBatch() { + System.out.println("Start test :" + TestHelper.getTestMethodName()); + final String clusterName = "TestDropInstanceSingleBatch"; + final String instanceName = "host_small"; + + HelixAdmin admin = new ZKHelixAdmin(_gZkClient); + admin.addCluster(clusterName, true); + admin.addInstance(clusterName, new InstanceConfig(instanceName)); + String msgPath = PropertyPathBuilder.instanceMessage(clusterName, instanceName); + for (int i = 0; i < 10; i++) { + _gZkClient.createPersistent(msgPath + "/msg-" + i); + } + + admin.dropInstance(clusterName, new InstanceConfig(instanceName)); + + AssertJUnit.assertFalse(_gZkClient.exists(PropertyPathBuilder.instance(clusterName, instanceName))); + _gSetupTool.deleteCluster(clusterName); + System.out.println("End test :" + TestHelper.getTestMethodName()); + } + + // Non-NotEmpty errors from multi() must NOT trigger the 3-retry loop. The + // production incident was 1880 threads stuck retrying CONNECTIONLOSS for 24h; + // we want fail-fast for anything that isn't the racy NotEmpty case. + @Test + public void testDropInstanceFailsFastOnNonRetryableMultiError() { + System.out.println("Start test :" + TestHelper.getTestMethodName()); + final String clusterName = "TestDropInstanceFailFast"; + final String instanceName = "host_failfast"; + InstanceConfig config = new InstanceConfig(instanceName); + + String instancePath = PropertyPathBuilder.instance(clusterName, instanceName); + String instanceConfigPath = PropertyPathBuilder.instanceConfig(clusterName, instanceName); + String liveInstancePath = PropertyPathBuilder.liveInstance(clusterName, instanceName); + + RealmAwareZkClient mockZkClient = Mockito.mock(RealmAwareZkClient.class); + Mockito.when(mockZkClient.exists(instanceConfigPath)).thenReturn(true); + Mockito.when(mockZkClient.exists(instancePath)).thenReturn(true); + Mockito.when(mockZkClient.exists(liveInstancePath)).thenReturn(false); + Mockito.when(mockZkClient.getChildren(instancePath)).thenReturn(Collections.emptyList()); + Mockito.when(mockZkClient.multi(Mockito.anyIterable())).thenReturn(Collections.singletonList( + (org.apache.zookeeper.OpResult) new org.apache.zookeeper.OpResult.ErrorResult( + KeeperException.Code.SYSTEMERROR.intValue()))); + + HelixAdmin helixAdminMock = new ZKHelixAdmin(mockZkClient); + long start = System.currentTimeMillis(); + try { + helixAdminMock.dropInstance(clusterName, config); + Assert.fail("Should throw HelixException"); + } catch (HelixException expected) { + // Should fail on the FIRST attempt - retryCnt=0 + Assert.assertEquals(expected.getMessage(), + "Failed to drop instance: " + instanceName + ". Retry times: 0", + "Non-NotEmpty errors must not trigger the 3-retry loop"); + } + long elapsed = System.currentTimeMillis() - start; + AssertJUnit.assertTrue("dropInstance should fail fast (took " + elapsed + " ms)", elapsed < 2000); + + // multi() should have been invoked exactly once (no retries) + Mockito.verify(mockZkClient, Mockito.times(1)).multi(Mockito.anyIterable()); + System.out.println("End test :" + TestHelper.getTestMethodName()); + } + + // multi() throwing (e.g. unrecoverable connection loss after lower-level + // ZkClient retries are exhausted) is wrapped as ZkClientException. The wrapped + // exception's cause is NOT the NotEmpty-shaped chain, so the outer retry loop + // must NOT retry - it must fail fast as HelixException("Retry times: 0"). + @Test + public void testDropInstanceFailsFastWhenMultiThrows() { + System.out.println("Start test :" + TestHelper.getTestMethodName()); + final String clusterName = "TestDropInstanceMultiThrows"; + final String instanceName = "host_multithrow"; + InstanceConfig config = new InstanceConfig(instanceName); + + String instancePath = PropertyPathBuilder.instance(clusterName, instanceName); + String instanceConfigPath = PropertyPathBuilder.instanceConfig(clusterName, instanceName); + String liveInstancePath = PropertyPathBuilder.liveInstance(clusterName, instanceName); + + RealmAwareZkClient mockZkClient = Mockito.mock(RealmAwareZkClient.class); + Mockito.when(mockZkClient.exists(instanceConfigPath)).thenReturn(true); + Mockito.when(mockZkClient.exists(instancePath)).thenReturn(true); + Mockito.when(mockZkClient.exists(liveInstancePath)).thenReturn(false); + Mockito.when(mockZkClient.getChildren(instancePath)).thenReturn(Collections.emptyList()); + Mockito.when(mockZkClient.multi(Mockito.anyIterable())) + .thenThrow(new RuntimeException("simulated unrecoverable ZK error")); + + HelixAdmin helixAdminMock = new ZKHelixAdmin(mockZkClient); + try { + helixAdminMock.dropInstance(clusterName, config); + Assert.fail("Should throw HelixException"); + } catch (HelixException expected) { + Assert.assertEquals(expected.getMessage(), + "Failed to drop instance: " + instanceName + ". Retry times: 0", + "multi() throws should not be retried by the outer NotEmpty loop"); + } + Mockito.verify(mockZkClient, Mockito.times(1)).multi(Mockito.anyIterable()); + System.out.println("End test :" + TestHelper.getTestMethodName()); + } }