ZOOKEEPER-4026: Support `OpCode.create2` in `OpCode.multi`

Currently, nesting `OpCode.create2` in `OpCode.multi` will not get `stat`(ZOOKEEPER-1297). The cause is multifold:
* `MultiOperationRecord.deserialize` decays `OpCode.create2` to  `OpCode.create`.
* `DataTree.processTxn` ignores `OpCode.create2`.

This commit complements ZOOKEEPER-1297 to add `stat` for `OpCode.create2` in `OpCode.multi`.

It also adds `CreateOptions` to cover create mode, acl and ttl to avoid massive overloading methods.

Author: Kezhu Wang <kezhuw@gmail.com>

Reviewers: Enrico Olivelli <eolivelli@apache.org>, Damien Diederen <ddiederen@apache.org>

Closes #1978 from kezhuw/ZOOKEEPER-4667-nest-create2-in-multi

zookeeper-server/src/main/java/org/apache/zookeeper/CreateOptions.java

@@ -0,0 +1,88 @@
+/*
+ * 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.
+ */
+
+package org.apache.zookeeper;
+
+import java.util.List;
+import java.util.Objects;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.server.EphemeralType;
+
+/**
+ * Options for creating znode in ZooKeeper data tree.
+ */
+public class CreateOptions {
+    private final CreateMode createMode;
+    private final List<ACL> acl;
+    private final long ttl;
+
+    public CreateMode getCreateMode() {
+        return createMode;
+    }
+
+    public List<ACL> getAcl() {
+        return acl;
+    }
+
+    public long getTtl() {
+        return ttl;
+    }
+
+    /**
+     * Constructs a builder for {@link CreateOptions}.
+     *
+     * @param acl
+     *                the acl for the node
+     * @param createMode
+     *                specifying whether the node to be created is ephemeral
+     *                and/or sequential
+     */
+    public static Builder newBuilder(List<ACL> acl, CreateMode createMode) {
+        return new Builder(createMode, acl);
+    }
+
+    private CreateOptions(CreateMode createMode, List<ACL> acl, long ttl) {
+        this.createMode = createMode;
+        this.acl = acl;
+        this.ttl = ttl;
+        EphemeralType.validateTTL(createMode, ttl);
+    }
+
+    /**
+     * Builder for {@link CreateOptions}.
+     */
+    public static class Builder {
+        private final CreateMode createMode;
+        private final List<ACL> acl;
+        private long ttl = -1;
+
+        private Builder(CreateMode createMode, List<ACL> acl) {
+            this.createMode = Objects.requireNonNull(createMode, "create mode is mandatory for create options");
+            this.acl = Objects.requireNonNull(acl, "acl is mandatory for create options");
+        }
+
+        public Builder withTtl(long ttl) {
+            this.ttl = ttl;
+            return this;
+        }
+
+        public CreateOptions build() {
+            return new CreateOptions(createMode, acl, ttl);
+        }
+    }
+}

zookeeper-server/src/main/java/org/apache/zookeeper/MultiOperationRecord.java

@@ -126,7 +126,12 @@ public class MultiOperationRecord implements Record, Iterable<Op> {
                 case ZooDefs.OpCode.createContainer:
                     CreateRequest cr = new CreateRequest();
                     cr.deserialize(archive, tag);
-                    add(Op.create(cr.getPath(), cr.getData(), cr.getAcl(), cr.getFlags()));
+                    CreateMode createMode = CreateMode.fromFlag(cr.getFlags(), null);
+                    if (createMode == null) {
+                        throw new IOException("invalid flag " + cr.getFlags() + " for create mode");
+                    }
+                    CreateOptions options = CreateOptions.newBuilder(cr.getAcl(), createMode).build();
+                    add(Op.create(cr.getPath(), cr.getData(), options, h.getType()));
                     break;
                 case ZooDefs.OpCode.createTTL:
                     CreateTTLRequest crTtl = new CreateTTLRequest();

zookeeper-server/src/main/java/org/apache/zookeeper/Op.java

@@ -152,6 +152,42 @@ public abstract class Op {
         return new Create(path, data, acl, createMode);
     }
 
+    /**
+     * Constructs a create operation which uses given op code if no one is inferred from create mode.
+     *
+     * @param path
+     *                the path for the node
+     * @param data
+     *                the initial data for the node
+     * @param options
+     *                options for creating znode
+     * @param defaultOpCode
+     *                op code to be used if no one is inferred from create mode
+     */
+    static Op create(String path, byte[] data, CreateOptions options, int defaultOpCode) {
+        if (options.getCreateMode().isTTL()) {
+            return new CreateTTL(path, data, options.getAcl(), options.getCreateMode(), options.getTtl());
+        }
+        return new Create(path, data, options.getAcl(), options.getCreateMode(), defaultOpCode);
+    }
+
+    /**
+     * Constructs a create operation which uses {@link ZooDefs.OpCode#create2} if no one is inferred from create mode.
+     *
+     * <p>The corresponding {@link OpResult.CreateResult#getStat()} could be null if connected to server without this
+     * patch.
+     *
+     * @param path
+     *                the path for the node
+     * @param data
+     *                the initial data for the node
+     * @param options
+     *                options for creating znode
+     */
+    public static Op create(String path, byte[] data, CreateOptions options) {
+        return create(path, data, options, ZooDefs.OpCode.create2);
+    }
+
     /**
      * Constructs a delete operation.  Arguments are as for the ZooKeeper method of the same name.
      * @see ZooKeeper#delete(String, int)
@@ -263,21 +299,29 @@ public abstract class Op {
         protected int flags;
 
         private Create(String path, byte[] data, List<ACL> acl, int flags) {
-            super(getOpcode(CreateMode.fromFlag(flags, CreateMode.PERSISTENT)), path, OpKind.TRANSACTION);
+            this(path, data, acl, flags, ZooDefs.OpCode.create);
+        }
+
+        private Create(String path, byte[] data, List<ACL> acl, int flags, int defaultOpCode) {
+            super(getOpcode(CreateMode.fromFlag(flags, CreateMode.PERSISTENT), defaultOpCode), path, OpKind.TRANSACTION);
             this.data = data;
             this.acl = acl;
             this.flags = flags;
         }
 
-        private static int getOpcode(CreateMode createMode) {
+        private static int getOpcode(CreateMode createMode, int defaultOpCode) {
             if (createMode.isTTL()) {
                 return ZooDefs.OpCode.createTTL;
             }
-            return createMode.isContainer() ? ZooDefs.OpCode.createContainer : ZooDefs.OpCode.create;
+            return createMode.isContainer() ? ZooDefs.OpCode.createContainer : defaultOpCode;
         }
 
         private Create(String path, byte[] data, List<ACL> acl, CreateMode createMode) {
-            super(getOpcode(createMode), path, OpKind.TRANSACTION);
+            this(path, data, acl, createMode, ZooDefs.OpCode.create);
+        }
+
+        private Create(String path, byte[] data, List<ACL> acl, CreateMode createMode, int defaultOpCode) {
+            super(getOpcode(createMode, defaultOpCode), path, OpKind.TRANSACTION);
             this.data = data;
             this.acl = acl;
             this.flags = createMode.toFlag();

zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java

@@ -973,6 +973,7 @@ public class DataTree {
                     Record record;
                     switch (subtxn.getType()) {
                     case OpCode.create:
+                    case OpCode.create2:
                         record = new CreateTxn();
                         break;
                     case OpCode.createTTL:

zookeeper-server/src/test/java/org/apache/zookeeper/MultiOperationRecordTest.java

@@ -33,7 +33,7 @@ public class MultiOperationRecordTest extends ZKTestCase {
     public void testRoundTrip() throws IOException {
         MultiOperationRecord request = new MultiOperationRecord();
         request.add(Op.check("check", 1));
-        request.add(Op.create("create", "create data".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL, ZooDefs.Perms.ALL));
+        request.add(Op.create("create", "create data".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.EPHEMERAL.toFlag()));
         request.add(Op.delete("delete", 17));
         request.add(Op.setData("setData", "set data".getBytes(), 19));
 

zookeeper-server/src/test/java/org/apache/zookeeper/server/CreateTTLTest.java

@@ -30,6 +30,7 @@ import java.util.List;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.zookeeper.AsyncCallback;
 import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.CreateOptions;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.Code;
 import org.apache.zookeeper.Op;
@@ -182,29 +183,46 @@ public class CreateTTLTest extends ClientBase {
 
     @Test
     public void testMulti() throws KeeperException, InterruptedException {
-        Op createTtl = Op.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_WITH_TTL, 100);
-        Op createTtlSequential = Op.create("/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL, 200);
+        CreateOptions options = CreateOptions
+                .newBuilder(ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_WITH_TTL)
+                .withTtl(100)
+                .build();
+        CreateOptions sequentialOptions = CreateOptions
+                .newBuilder(ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL)
+                .withTtl(200)
+                .build();
+        Op createTtl = Op.create("/a", new byte[0], options.getAcl(), options.getCreateMode(), options.getTtl());
+        Op createTtl2 = Op.create("/a2", new byte[0], options);
+        Op createTtlSequential = Op.create("/b", new byte[0], sequentialOptions.getAcl(), sequentialOptions.getCreateMode(), sequentialOptions.getTtl());
+        Op createTtlSequential2 = Op.create("/b2", new byte[0], sequentialOptions);
         Op createNonTtl = Op.create("/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-        List<OpResult> results = zk.multi(Arrays.asList(createTtl, createTtlSequential, createNonTtl));
-        String sequentialPath = ((OpResult.CreateResult) results.get(1)).getPath();
+        List<OpResult> results = zk.multi(Arrays.asList(createTtl, createTtl2, createTtlSequential, createTtlSequential2, createNonTtl));
+        String sequentialPath = ((OpResult.CreateResult) results.get(2)).getPath();
+        String sequentialPath2 = ((OpResult.CreateResult) results.get(3)).getPath();
 
         final AtomicLong fakeElapsed = new AtomicLong(0);
         ContainerManager containerManager = newContainerManager(fakeElapsed);
         containerManager.checkContainers();
         assertNotNull(zk.exists("/a", false), "node should not have been deleted yet");
+        assertNotNull(zk.exists("/a2", false), "node should not have been deleted yet");
         assertNotNull(zk.exists(sequentialPath, false), "node should not have been deleted yet");
+        assertNotNull(zk.exists(sequentialPath2, false), "node should not have been deleted yet");
         assertNotNull(zk.exists("/c", false), "node should never be deleted");
 
         fakeElapsed.set(110);
         containerManager.checkContainers();
         assertNull(zk.exists("/a", false), "node should have been deleted");
+        assertNull(zk.exists("/a2", false), "node should have been deleted");
         assertNotNull(zk.exists(sequentialPath, false), "node should not have been deleted yet");
+        assertNotNull(zk.exists(sequentialPath2, false), "node should not have been deleted yet");
         assertNotNull(zk.exists("/c", false), "node should never be deleted");
 
         fakeElapsed.set(210);
         containerManager.checkContainers();
         assertNull(zk.exists("/a", false), "node should have been deleted");
+        assertNull(zk.exists("/a2", false), "node should have been deleted");
         assertNull(zk.exists(sequentialPath, false), "node should have been deleted");
+        assertNull(zk.exists(sequentialPath2, false), "node should have been deleted");
         assertNotNull(zk.exists("/c", false), "node should never be deleted");
     }
 

zookeeper-server/src/test/java/org/apache/zookeeper/test/MultiOperationTest.java

@@ -21,6 +21,7 @@ package org.apache.zookeeper.test;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
@@ -40,6 +41,7 @@ import org.apache.zookeeper.AsyncCallback;
 import org.apache.zookeeper.AsyncCallback.MultiCallback;
 import org.apache.zookeeper.ClientCnxn;
 import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.CreateOptions;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.Op;
 import org.apache.zookeeper.OpResult;
@@ -443,6 +445,28 @@ public class MultiOperationTest extends ClientBase {
         zk.getData("/multi2", false, null);
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testCreate2(boolean useAsync) throws Exception {
+        CreateOptions options = CreateOptions.newBuilder(Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT).build();
+        List<Op> ops = Arrays.asList(
+            Op.create("/multi0", new byte[0], options),
+            Op.create("/multi1", new byte[0], options),
+            Op.create("/multi2", new byte[0], options));
+        List<OpResult> results = multi(zk, ops, useAsync);
+        for (int i = 0; i < ops.size(); i++) {
+            CreateResult createResult = (CreateResult) results.get(i);
+            assertEquals(ops.get(i).getPath(), createResult.getPath());
+            assertEquals(ZooDefs.OpCode.create2, createResult.getType(), createResult.getPath());
+            assertNotNull(createResult.getStat(), createResult.getPath());
+            assertNotEquals(0, createResult.getStat().getCzxid(), createResult.getPath());
+        }
+
+        zk.getData("/multi0", false, null);
+        zk.getData("/multi1", false, null);
+        zk.getData("/multi2", false, null);
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testEmpty(boolean useAsync) throws Exception {