ZOOKEEPER-2052 Unable to delete a node when the node has no children (Hongchao Deng and Yip Ng via rakeshr)

git-svn-id: https://svn.apache.org/repos/asf/zookeeper/trunk@1634776 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -12,6 +12,9 @@ BUGFIXES:
   ZOOKEEPER-2049 Yosemite build failure: htonll conflict (Till Toenshoff via
   michim)
 
+  ZOOKEEPER-2052 Unable to delete a node when the node has no children
+  (Hongchao Deng and Yip Ng via rakeshr)
+
 IMPROVEMENTS:
   ZOOKEEPER-1660 Documentation for Dynamic Reconfiguration (Reed Wanderman-Milne via shralex)  
 

src/java/main/org/apache/zookeeper/server/PrepRequestProcessor.java

@@ -159,14 +159,6 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
         ChangeRecord lastChange = null;
         synchronized (zks.outstandingChanges) {
             lastChange = zks.outstandingChangesForPath.get(path);
-            /*
-            for (int i = 0; i < zks.outstandingChanges.size(); i++) {
-                ChangeRecord c = zks.outstandingChanges.get(i);
-                if (c.path.equals(path)) {
-                    lastChange = c;
-                }
-            }
-            */
             if (lastChange == null) {
                 DataNode n = zks.getZKDatabase().getNode(path);
                 if (n != null) {
@@ -188,6 +180,12 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
         return lastChange;
     }
 
+    private ChangeRecord getOutstandingChange(String path) {
+        synchronized (zks.outstandingChanges) {
+            return zks.outstandingChangesForPath.get(path);
+        }
+    }
+
     private void addChangeRecord(ChangeRecord c) {
         synchronized (zks.outstandingChanges) {
             zks.outstandingChanges.add(c);
@@ -202,39 +200,37 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
      * of a failed multi-op.
      *
      * @param multiRequest
+     * @return a map that contains previously existed records that probably need to be
+     *         rolled back in any failure.
      */
     private Map<String, ChangeRecord> getPendingChanges(MultiTransactionRecord multiRequest) {
         HashMap<String, ChangeRecord> pendingChangeRecords = new HashMap<String, ChangeRecord>();
 
-        for(Op op: multiRequest) {
+        for (Op op : multiRequest) {
             String path = op.getPath();
+            ChangeRecord cr = getOutstandingChange(path);
+            // only previously existing records need to be rolled back.
+            if (cr != null) {
+                pendingChangeRecords.put(path, cr);
+            }
 
-            try {
-                ChangeRecord cr = getRecordForPath(path);
-                if (cr != null) {
-                    pendingChangeRecords.put(path, cr);
-                }
-
-                /*
-                 * ZOOKEEPER-1624 - We need to store for parent's ChangeRecord
-                 * of the parent node of a request. So that if this is a
-                 * sequential node creation request, rollbackPendingChanges()
-                 * can restore previous parent's ChangeRecord correctly.
-                 *
-                 * Otherwise, sequential node name generation will be incorrect
-                 * for a subsequent request.
-                 */
-                int lastSlash = path.lastIndexOf('/');
-                if (lastSlash == -1 || path.indexOf('\0') != -1) {
-                    continue;
-                }
-                String parentPath = path.substring(0, lastSlash);
-                ChangeRecord parentCr = getRecordForPath(parentPath);
-                if (parentCr != null) {
-                    pendingChangeRecords.put(parentPath, parentCr);
-                }
-            } catch (KeeperException.NoNodeException e) {
-                // ignore this one
+            /*
+             * ZOOKEEPER-1624 - We need to store for parent's ChangeRecord
+             * of the parent node of a request. So that if this is a
+             * sequential node creation request, rollbackPendingChanges()
+             * can restore previous parent's ChangeRecord correctly.
+             *
+             * Otherwise, sequential node name generation will be incorrect
+             * for a subsequent request.
+             */
+            int lastSlash = path.lastIndexOf('/');
+            if (lastSlash == -1 || path.indexOf('\0') != -1) {
+                continue;
+            }
+            String parentPath = path.substring(0, lastSlash);
+            ChangeRecord parentCr = getOutstandingChange(parentPath);
+            if (parentCr != null) {
+                pendingChangeRecords.put(parentPath, parentCr);
             }
         }
 
@@ -252,7 +248,6 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
      * @param pendingChangeRecords
      */
     void rollbackPendingChanges(long zxid, Map<String, ChangeRecord>pendingChangeRecords) {
-
         synchronized (zks.outstandingChanges) {
             // Grab a list iterator starting at the END of the list so we can iterate in reverse
             ListIterator<ChangeRecord> iter = zks.outstandingChanges.listIterator(zks.outstandingChanges.size());
@@ -260,27 +255,30 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 ChangeRecord c = iter.previous();
                 if (c.zxid == zxid) {
                     iter.remove();
+                    // Remove all outstanding changes for paths of this multi.
+                    // Previous records will be added back later.
                     zks.outstandingChangesForPath.remove(c.path);
                 } else {
                     break;
                 }
             }
 
-            boolean empty = zks.outstandingChanges.isEmpty();
-            long firstZxid = 0;
-            if (!empty) {
-                firstZxid = zks.outstandingChanges.get(0).zxid;
+            // we don't need to roll back any records because there is nothing left.
+            if (zks.outstandingChanges.isEmpty()) {
+                return;
             }
 
-            Iterator<ChangeRecord> priorIter = pendingChangeRecords.values().iterator();
-            while (priorIter.hasNext()) {
-                ChangeRecord c = priorIter.next();
+            long firstZxid = zks.outstandingChanges.get(0).zxid;
 
-                /* Don't apply any prior change records less than firstZxid */
-                if (!empty && (c.zxid < firstZxid)) {
+            for (ChangeRecord c : pendingChangeRecords.values()) {
+                // Don't apply any prior change records less than firstZxid.
+                // Note that previous outstanding requests might have been removed
+                // once they are completed.
+                if (c.zxid < firstZxid) {
                     continue;
                 }
 
+                // add previously existing records back.
                 zks.outstandingChangesForPath.put(c.path, c);
             }
         }

src/java/test/org/apache/zookeeper/server/PrepRequestProcessorTest.java

@@ -18,63 +18,181 @@
 
 package org.apache.zookeeper.server;
 
-import static org.junit.Assert.*;
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.jute.Record;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.SessionExpiredException;
+import org.apache.zookeeper.KeeperException.SessionMovedException;
+import org.apache.zookeeper.MultiTransactionRecord;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooDefs.OpCode;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.ZooKeeperServer.ChangeRecord;
+import org.apache.zookeeper.test.ClientBase;
+import org.apache.zookeeper.txn.ErrorTxn;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
+import java.io.IOException;
 import java.io.PrintWriter;
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 
-import org.apache.zookeeper.PortAssignment;
-import org.apache.zookeeper.KeeperException.Code;
-import org.apache.zookeeper.KeeperException.SessionExpiredException;
-import org.apache.zookeeper.KeeperException.SessionMovedException;
-import org.apache.zookeeper.ZooDefs.OpCode;
-import org.apache.zookeeper.server.PrepRequestProcessor;
-import org.apache.zookeeper.server.Request;
-import org.apache.zookeeper.server.RequestProcessor;
-import org.apache.zookeeper.server.ServerCnxnFactory;
-import org.apache.zookeeper.server.SyncRequestProcessor;
-import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.test.ClientBase;
-import org.apache.zookeeper.txn.ErrorTxn;
-import org.junit.Assert;
-import org.junit.Test;
-
 public class PrepRequestProcessorTest extends ClientBase {
-    private static String HOSTPORT = "127.0.0.1:" + PortAssignment.unique();
+    private static final Logger LOG = LoggerFactory.getLogger(PrepRequestProcessorTest.class);
     private static final int CONNECTION_TIMEOUT = 3000;
-    private final CountDownLatch testEnd = new CountDownLatch(1);
+    private static String HOSTPORT = "127.0.0.1:" + PortAssignment.unique();
+    private CountDownLatch pLatch;
 
-    @Test
-    public void testPRequest() throws Exception {
+    private ZooKeeperServer zks;
+    private ServerCnxnFactory servcnxnf;
+    private PrepRequestProcessor processor;
+    private Request outcome;
+
+    @Before
+    public void setup() throws Exception {
         File tmpDir = ClientBase.createTmpDir();
         ClientBase.setupTestEnv();
-        ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
+        zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
         SyncRequestProcessor.setSnapCount(100);
         final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        ServerCnxnFactory f = ServerCnxnFactory.createFactory(PORT, -1);
-        f.startup(zks);
+
+        servcnxnf = ServerCnxnFactory.createFactory(PORT, -1);
+        servcnxnf.startup(zks);
         Assert.assertTrue("waiting for server being up ",
-                ClientBase.waitForServerUp(HOSTPORT,CONNECTION_TIMEOUT));
-        zks.sessionTracker = new MySessionTracker(); 
-        PrepRequestProcessor processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
+                ClientBase.waitForServerUp(HOSTPORT, CONNECTION_TIMEOUT));
+        zks.sessionTracker = new MySessionTracker();
+    }
+
+    @After
+    public void teardown() throws Exception {
+        if (servcnxnf != null) {
+            servcnxnf.shutdown();
+        }
+        if (zks != null) {
+            zks.shutdown();
+        }
+    }
+
+    @Test
+    public void testPRequest() throws Exception {
+        pLatch = new CountDownLatch(1);
+        processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
         Request foo = new Request(null, 1l, 1, OpCode.create, ByteBuffer.allocate(3), null);
         processor.pRequest(foo);
-        testEnd.await(5, java.util.concurrent.TimeUnit.SECONDS);
-        f.shutdown();
-        zks.shutdown();
+
+        Assert.assertEquals("Request should have marshalling error", new ErrorTxn(KeeperException.Code.MARSHALLINGERROR.intValue()),
+                outcome.getTxn());
+        Assert.assertTrue("request hasn't been processed in chain",
+                pLatch.await(5, java.util.concurrent.TimeUnit.SECONDS));
+    }
+
+    private Request createMultiRequest(List<Op> ops) throws IOException {
+        Record record = new MultiTransactionRecord(ops);
+
+        // encoding
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+        record.serialize(boa, "request");
+        baos.close();
+
+        // Id
+        List<Id> ids = Arrays.asList(Ids.ANYONE_ID_UNSAFE);
+
+        return new Request(null, 1l, 0, OpCode.multi, ByteBuffer.wrap(baos.toByteArray()), ids);
+    }
+
+    private void process(List<Op> ops) throws Exception {
+        pLatch = new CountDownLatch(1);
+        processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
+
+        Request req = createMultiRequest(ops);
+
+        processor.pRequest(req);
+        Assert.assertTrue("request hasn't been processed in chain",
+                pLatch.await(5, java.util.concurrent.TimeUnit.SECONDS));
+    }
+
+    /**
+     * This test checks that a successful multi will change outstanding record
+     * and failed multi shouldn't change outstanding record.
+     */
+    @Test
+    public void testMultiOutstandingChange() throws Exception {
+        zks.getZKDatabase().dataTree.createNode("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, 0, 0, 0, 0);
+
+        Assert.assertNull(zks.outstandingChangesForPath.get("/foo"));
+
+        process(Arrays.asList(
+                Op.setData("/foo", new byte[0], -1)));
+
+        ChangeRecord cr = zks.outstandingChangesForPath.get("/foo");
+        Assert.assertNotNull("Change record wasn't set", cr);
+        Assert.assertEquals("Record zxid wasn't set correctly",
+                1, cr.zxid);
+
+        process(Arrays.asList(
+                Op.delete("/foo", -1)));
+        cr = zks.outstandingChangesForPath.get("/foo");
+        Assert.assertEquals("Record zxid wasn't set correctly",
+                2, cr.zxid);
+
+
+        // It should fail and shouldn't change outstanding record.
+        process(Arrays.asList(
+                Op.delete("/foo", -1)));
+        cr = zks.outstandingChangesForPath.get("/foo");
+        // zxid should still be previous result because record's not changed.
+        Assert.assertEquals("Record zxid wasn't set correctly",
+                2, cr.zxid);
+    }
+
+    /**
+     * ZOOKEEPER-2052:
+     * This test checks that if a multi operation aborted, and during the multi there is side effect
+     * that changed outstandingChangesForPath, after aborted the side effect should be removed and
+     * everything should be restored correctly.
+     */
+    @Test
+    public void testMultiRollbackNoLastChange() throws Exception {
+        zks.getZKDatabase().dataTree.createNode("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, 0, 0, 0, 0);
+        zks.getZKDatabase().dataTree.createNode("/foo/bar", new byte[0], Ids.OPEN_ACL_UNSAFE, 0, 0, 0, 0);
+
+        pLatch = new CountDownLatch(1);
+        processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
+
+        Assert.assertNull(zks.outstandingChangesForPath.get("/foo"));
+
+        // multi record:
+        //   set "/foo" => succeed, leave a outstanding change
+        //   delete "/foo" => fail, roll back change
+        process(Arrays.asList(
+                Op.setData("/foo", new byte[0], -1),
+                Op.delete("/foo", -1)));
+
+        // aborting multi shouldn't leave any record.
+        Assert.assertNull(zks.outstandingChangesForPath.get("/foo"));
     }
- 
 
     private class MyRequestProcessor implements RequestProcessor {
         @Override
         public void processRequest(Request request) {
-            Assert.assertEquals("Request should have marshalling error", new ErrorTxn(Code.MARSHALLINGERROR.intValue()),  request.getTxn());
-            testEnd.countDown();
+            // getting called by PrepRequestProcessor
+            outcome = request;
+            pLatch.countDown();
         }
         @Override
         public void shutdown() {

src/java/test/org/apache/zookeeper/test/MultiTransactionTest.java

@@ -1,10 +1,11 @@
-/*
- * 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
+/**
+ * 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
  *
@@ -241,6 +242,58 @@ public class MultiTransactionTest extends ClientBase {
         multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
     }
 
+    /**
+     * ZOOKEEPER-2052:
+     * Multi abort shouldn't have any side effect.
+     * We fix a bug in rollback and the following scenario should work:
+     * 1. multi delete abort because of not empty directory
+     * 2. ephemeral nodes under that directory are deleted
+     * 3. multi delete should succeed.
+     */
+    @Test
+    public void testMultiRollback() throws Exception {
+        zk.create("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+        ZooKeeper epheZk = createClient();
+        epheZk.create("/foo/bar", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+
+        List<Op> opList = Arrays.asList(Op.delete("/foo", -1));
+        try {
+            zk.multi(opList);
+            Assert.fail("multi delete should failed for not empty directory");
+        } catch (KeeperException.NotEmptyException e) {
+        }
+
+        final CountDownLatch latch = new CountDownLatch(1);
+
+        zk.exists("/foo/bar", new Watcher() {
+            @Override
+            public void process(WatchedEvent event) {
+                if (event.getType() == Event.EventType.NodeDeleted){
+                    latch.countDown();
+                }
+            }
+        });
+
+        epheZk.close();
+
+        latch.await();
+
+        try {
+            zk.getData("/foo/bar", false, null);
+            Assert.fail("ephemeral node should have been deleted");
+        } catch (KeeperException.NoNodeException e) {
+        }
+
+        zk.multi(opList);
+
+        try {
+            zk.getData("/foo", false, null);
+            Assert.fail("persistent node should have been deleted after multi");
+        } catch (KeeperException.NoNodeException e) {
+        }
+    }
+
     /**
      * Test verifies the multi calls with blank znode path
      */