From 3682d69a31b72f5bc2b5370e77ae83afa4b61682 Mon Sep 17 00:00:00 2001
From: Sylvain Wallez <sylvain@bluxte.net>
Date: Mon, 16 Sep 2019 15:27:47 +0200
Subject: [PATCH] Delete empty containers with cversion == 0 after a grace
 period

---
 .../zookeeper/server/ContainerManager.java    | 25 +++++++++++++------
 .../zookeeper/server/CreateContainerTest.java | 16 ++++++++++++
 2 files changed, 33 insertions(+), 8 deletions(-)

diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ContainerManager.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ContainerManager.java
index f6f0e973d45..dea56a969b4 100644
--- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ContainerManager.java
+++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ContainerManager.java
@@ -48,6 +48,7 @@ public class ContainerManager {
     private final int maxPerMinute;
     private final Timer timer;
     private final AtomicReference<TimerTask> task = new AtomicReference<TimerTask>(null);
+    private final Set<String> noChildrenAtLastCheck = new HashSet<String>();
 
     /**
      * @param zkDb the ZK database
@@ -139,14 +140,22 @@ protected Collection<String> getCandidates() {
         Set<String> candidates = new HashSet<String>();
         for (String containerPath : zkDb.getDataTree().getContainers()) {
             DataNode node = zkDb.getDataTree().getNode(containerPath);
-            /*
-                cversion > 0: keep newly created containers from being deleted
-                before any children have been added. If you were to create the
-                container just before a container cleaning period the container
-                would be immediately be deleted.
-             */
-            if ((node != null) && (node.stat.getCversion() > 0) && (node.getChildren().isEmpty())) {
-                candidates.add(containerPath);
+            boolean wasNewWithNoChildren = noChildrenAtLastCheck.remove(containerPath);
+
+            if (node != null && node.getChildren().isEmpty()) {
+                if (node.stat.getCversion() == 0) {
+                    // Give newly created containers a grace period and avoid deleting
+                    // them before any children could be added. If you were to create the
+                    // container just before a container cleaning period the container
+                    // would be immediately be deleted.
+                    if (wasNewWithNoChildren) {
+                        candidates.add(containerPath);
+                    } else {
+                        noChildrenAtLastCheck.add(containerPath);
+                    }
+                } else {
+                    candidates.add(containerPath);
+                }
             }
         }
         for (String ttlPath : zkDb.getDataTree().getTtls()) {
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/CreateContainerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/CreateContainerTest.java
index 83c7f0b3db9..41ccf7b3ed7 100644
--- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/CreateContainerTest.java
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/CreateContainerTest.java
@@ -237,6 +237,22 @@ public Void call() throws Exception {
         assertEquals(queue.poll(5, TimeUnit.SECONDS), "/four");
     }
 
+    @Test(timeout = 30000)
+    public void testContainerWithNoChildGracePeriod() throws KeeperException, InterruptedException {
+        zk.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.CONTAINER);
+
+        ContainerManager containerManager = new ContainerManager(serverFactory.getZooKeeperServer().getZKDatabase(), serverFactory.getZooKeeperServer().firstProcessor, 1, 100);
+        containerManager.checkContainers();
+        Thread.sleep(1000);
+
+        assertNotNull("Container should still be there", zk.exists("/foo", false));
+
+        containerManager.checkContainers();
+        Thread.sleep(1000);
+
+        assertNull("Container should have been deleted", zk.exists("/foo", false));
+    }
+
     private void createNoStatVerifyResult(String newName) throws KeeperException, InterruptedException {
         assertNull("Node existed before created", zk.exists(newName, false));
         zk.create(newName, newName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.CONTAINER);