ZOOKEEPER-2169. Enable creation of nodes with TTLs. (Jordan Zimmerman via camille)

CHANGES.txt

@@ -14,6 +14,8 @@ NEW FEATURES:
 
   ZOOKEEPER-1962: Add a CLI command to recursively list a znode and
   children (Gautam Gopalakrishnan, Hongchao Deng, Enis Soztutar via phunt)
+  
+  ZOOKEEPER-2169: Enable creation of nodes with TTLs (Jordan Zimmerman via camille)
 
 BUGFIXES:
   ZOOKEEPER-1784 wrong check for COMMITANDACTIVATE in observer code, Learner.java (rgs via shralex).

docs/zookeeperAdmin.html

@@ -1787,7 +1787,7 @@ server.3=zoo3:2888:3888</pre>
 <p>
 <strong>New in 3.6.0:</strong> The
                 time interval in milliseconds for each check of candidate container
-                nodes. Default is "60000".</p>
+                and ttl nodes. Default is "60000".</p>
 </dd>
 
           
@@ -1798,7 +1798,7 @@ server.3=zoo3:2888:3888</pre>
 <p>(Java system property only)</p>
 <p>
 <strong>New in 3.6.0:</strong> The
-                maximum number of container nodes that can be deleted per
+                maximum number of container and ttl nodes that can be deleted per
                 minute. This prevents herding during container deletion.
                 Default is "10000".</p>
 </dd>

docs/zookeeperProgrammers.html

@@ -219,6 +219,9 @@ document.write("Last Published: " + document.lastModified);
 <li>
 <a href="#Container+Nodes">Container Nodes</a>
 </li>
+<li>
+<a href="#TTL+Nodes">TTL Nodes</a>
+</li>
 </ul>
 </li>
 <li>
@@ -575,15 +578,24 @@ document.write("Last Published: " + document.lastModified);
 <p>
 <strong>Added in 3.6.0</strong>
 </p>
-<p>ZooKeeper has the notion of container nodes. Container nodes are
-          special purpose nodes useful for recipes such as leader, lock, etc.
+<p>ZooKeeper has the notion of container znodes. Container znodes are
+          special purpose znodes useful for recipes such as leader, lock, etc.
           When the last child of a container is deleted, the container becomes
           a candidate to be deleted by the server at some point in the future.</p>
 <p>Given this property, you should be prepared to get
           KeeperException.NoNodeException when creating children inside of
-          container nodes. i.e. when creating child nodes inside of container nodes
+          container znodes. i.e. when creating child znodes inside of container znodes
           always check for KeeperException.NoNodeException and recreate the container
-          node when it occurs.</p>
+          znode when it occurs.</p>
+<a name="TTL+Nodes"></a>
+<h4>TTL Nodes</h4>
+<p>
+<strong>Added in 3.6.0</strong>
+</p>
+<p>When creating PERSISTENT or PERSISTENT_SEQUENTIAL znodes,
+          you can optionally set a TTL in milliseconds for the znode. If the znode
+          is not modified within the TTL and has no children it will become a candidate
+          to be deleted by the server at some point in the future.</p>
 <a name="sc_timeInZk"></a>
 <h3 class="h4">Time in ZooKeeper</h3>
 <p>ZooKeeper tracks time multiple ways:</p>

src/docs/src/documentation/content/xdocs/zookeeperAdmin.xml

@@ -1439,7 +1439,7 @@ server.3=zoo3:2888:3888</programlisting>
 
               <para><emphasis role="bold">New in 3.6.0:</emphasis> The
                 time interval in milliseconds for each check of candidate container
-                nodes. Default is "60000".</para>
+                and ttl nodes. Default is "60000".</para>
             </listitem>
           </varlistentry>
 
@@ -1450,7 +1450,7 @@ server.3=zoo3:2888:3888</programlisting>
               <para>(Java system property only)</para>
 
               <para><emphasis role="bold">New in 3.6.0:</emphasis> The
-                maximum number of container nodes that can be deleted per
+                maximum number of container and ttl nodes that can be deleted per
                 minute. This prevents herding during container deletion.
                 Default is "10000".</para>
             </listitem>

src/docs/src/documentation/content/xdocs/zookeeperProgrammers.xml

@@ -249,16 +249,27 @@
 
         <para><emphasis role="bold">Added in 3.6.0</emphasis></para>
 
-        <para>ZooKeeper has the notion of container nodes. Container nodes are
-          special purpose nodes useful for recipes such as leader, lock, etc.
+        <para>ZooKeeper has the notion of container znodes. Container znodes are
+          special purpose znodes useful for recipes such as leader, lock, etc.
           When the last child of a container is deleted, the container becomes
           a candidate to be deleted by the server at some point in the future.</para>
 
         <para>Given this property, you should be prepared to get
           KeeperException.NoNodeException when creating children inside of
-          container nodes. i.e. when creating child nodes inside of container nodes
+          container znodes. i.e. when creating child znodes inside of container znodes
           always check for KeeperException.NoNodeException and recreate the container
-          node when it occurs.</para>
+          znode when it occurs.</para>
+      </section>
+
+      <section>
+        <title>TTL Nodes</title>
+
+        <para><emphasis role="bold">Added in 3.6.0</emphasis></para>
+
+        <para>When creating PERSISTENT or PERSISTENT_SEQUENTIAL znodes,
+          you can optionally set a TTL in milliseconds for the znode. If the znode
+          is not modified within the TTL and has no children it will become a candidate
+          to be deleted by the server at some point in the future.</para>
       </section>
     </section>
 

src/java/main/org/apache/zookeeper/CreateMode.java

@@ -28,21 +28,21 @@ public enum CreateMode {
     /**
      * The znode will not be automatically deleted upon client's disconnect.
      */
-    PERSISTENT (0, false, false, false),
+    PERSISTENT (0, false, false, false, false),
     /**
     * The znode will not be automatically deleted upon client's disconnect,
     * and its name will be appended with a monotonically increasing number.
     */
-    PERSISTENT_SEQUENTIAL (2, false, true, false),
+    PERSISTENT_SEQUENTIAL (2, false, true, false, false),
     /**
      * The znode will be deleted upon the client's disconnect.
      */
-    EPHEMERAL (1, true, false, false),
+    EPHEMERAL (1, true, false, false, false),
     /**
      * The znode will be deleted upon the client's disconnect, and its name
      * will be appended with a monotonically increasing number.
      */
-    EPHEMERAL_SEQUENTIAL (3, true, true, false),
+    EPHEMERAL_SEQUENTIAL (3, true, true, false, false),
     /**
      * The znode will be a container node. Container
      * nodes are special purpose nodes useful for recipes such as leader, lock,
@@ -52,7 +52,20 @@ public enum CreateMode {
      * {@link org.apache.zookeeper.KeeperException.NoNodeException}
      * when creating children inside of this container node.
      */
-    CONTAINER (4, false, false, true);
+    CONTAINER (4, false, false, true, false),
+    /**
+     * The znode will not be automatically deleted upon client's disconnect.
+     * However if the znode has not been modified within the given TTL, it
+     * will be deleted once it has no children.
+     */
+    PERSISTENT_WITH_TTL(5, false, false, false, true),
+    /**
+     * The znode will not be automatically deleted upon client's disconnect,
+     * and its name will be appended with a monotonically increasing number.
+     * However if the znode has not been modified within the given TTL, it
+     * will be deleted once it has no children.
+     */
+    PERSISTENT_SEQUENTIAL_WITH_TTL(6, false, true, false, true);
 
     private static final Logger LOG = LoggerFactory.getLogger(CreateMode.class);
 
@@ -60,13 +73,15 @@ public enum CreateMode {
     private boolean sequential;
     private final boolean isContainer;
     private int flag;
+    private boolean isTTL;
 
     CreateMode(int flag, boolean ephemeral, boolean sequential,
-               boolean isContainer) {
+               boolean isContainer, boolean isTTL) {
         this.flag = flag;
         this.ephemeral = ephemeral;
         this.sequential = sequential;
         this.isContainer = isContainer;
+        this.isTTL = isTTL;
     }
 
     public boolean isEphemeral() { 
@@ -81,6 +96,10 @@ public enum CreateMode {
         return isContainer;
     }
 
+    public boolean isTTL() {
+        return isTTL;
+    }
+
     public int toFlag() {
         return flag;
     }
@@ -100,6 +119,10 @@ public enum CreateMode {
 
         case 4: return CreateMode.CONTAINER;
 
+        case 5: return CreateMode.PERSISTENT_WITH_TTL;
+
+        case 6: return CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL;
+
         default:
             String errMsg = "Received an invalid flag value: " + flag
                     + " to convert to a CreateMode";
@@ -128,6 +151,12 @@ public enum CreateMode {
             case 4:
                 return CreateMode.CONTAINER;
 
+            case 5:
+                return CreateMode.PERSISTENT_WITH_TTL;
+
+            case 6:
+                return CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL;
+
             default:
                 return defaultMode;
         }

src/java/main/org/apache/zookeeper/MultiTransactionRecord.java

@@ -68,6 +68,7 @@ public class MultiTransactionRecord implements Record, Iterable<Op> {
             switch (op.getType()) {
                 case ZooDefs.OpCode.create:
                 case ZooDefs.OpCode.create2:
+                case ZooDefs.OpCode.createTTL:
                 case ZooDefs.OpCode.createContainer:
                 case ZooDefs.OpCode.delete:
                 case ZooDefs.OpCode.setData:
@@ -97,6 +98,11 @@ public class MultiTransactionRecord implements Record, Iterable<Op> {
                     cr.deserialize(archive, tag);
                     add(Op.create(cr.getPath(), cr.getData(), cr.getAcl(), cr.getFlags()));
                     break;
+                case ZooDefs.OpCode.createTTL:
+                    CreateTTLRequest crTtl = new CreateTTLRequest();
+                    crTtl.deserialize(archive, tag);
+                    add(Op.create(crTtl.getPath(), crTtl.getData(), crTtl.getAcl(), crTtl.getFlags(), crTtl.getTtl()));
+                    break;
                 case ZooDefs.OpCode.delete:
                     DeleteRequest dr = new DeleteRequest();
                     dr.deserialize(archive, tag);

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

@@ -22,8 +22,10 @@ import org.apache.zookeeper.common.PathUtils;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.proto.CheckVersionRequest;
 import org.apache.zookeeper.proto.CreateRequest;
+import org.apache.zookeeper.proto.CreateTTLRequest;
 import org.apache.zookeeper.proto.DeleteRequest;
 import org.apache.zookeeper.proto.SetDataRequest;
+import org.apache.zookeeper.server.EphemeralType;
 
 import java.util.Arrays;
 import java.util.Iterator;
@@ -70,6 +72,32 @@ public abstract class Op {
         return new Create(path, data, acl, flags);
     }
 
+    /**
+     * Constructs a create operation.  Arguments are as for the ZooKeeper method of the same name
+     * but adding an optional ttl
+     * @see ZooKeeper#create(String, byte[], java.util.List, CreateMode)
+     * @see CreateMode#fromFlag(int)
+     *
+     * @param path
+     *                the path for the node
+     * @param data
+     *                the initial data for the node
+     * @param acl
+     *                the acl for the node
+     * @param flags
+     *                specifying whether the node to be created is ephemeral
+     *                and/or sequential but using the integer encoding.
+     * @param ttl
+     *                optional ttl or 0 (flags must imply a TTL creation mode)
+     */
+    public static Op create(String path, byte[] data, List<ACL> acl, int flags, long ttl) {
+        CreateMode createMode = CreateMode.fromFlag(flags, CreateMode.PERSISTENT);
+        if (createMode.isTTL()) {
+            return new CreateTTL(path, data, acl, createMode, ttl);
+        }
+        return new Create(path, data, acl, flags);
+    }
+
     /**
      * Constructs a create operation.  Arguments are as for the ZooKeeper method of the same name.
      * @see ZooKeeper#create(String, byte[], java.util.List, CreateMode)
@@ -88,6 +116,30 @@ public abstract class Op {
         return new Create(path, data, acl, createMode);
     }
 
+    /**
+     * Constructs a create operation.  Arguments are as for the ZooKeeper method of the same name
+     * but adding an optional ttl
+     * @see ZooKeeper#create(String, byte[], java.util.List, CreateMode)
+     *
+     * @param path
+     *                the path for the node
+     * @param data
+     *                the initial data for the node
+     * @param acl
+     *                the acl for the node
+     * @param createMode
+     *                specifying whether the node to be created is ephemeral
+     *                and/or sequential
+     * @param ttl
+     *                optional ttl or 0 (createMode must imply a TTL)
+     */
+    public static Op create(String path, byte[] data, List<ACL> acl, CreateMode createMode, long ttl) {
+        if (createMode.isTTL()) {
+            return new CreateTTL(path, data, acl, createMode, ttl);
+        }
+        return new Create(path, data, acl, createMode);
+    }
+
     /**
      * Constructs a delete operation.  Arguments are as for the ZooKeeper method of the same name.
      * @see ZooKeeper#delete(String, int)
@@ -178,9 +230,9 @@ public abstract class Op {
     // these internal classes are public, but should not generally be referenced.
     //
     public static class Create extends Op {
-        private byte[] data;
-        private List<ACL> acl;
-        private int flags;
+        protected byte[] data;
+        protected List<ACL> acl;
+        protected int flags;
 
         private Create(String path, byte[] data, List<ACL> acl, int flags) {
             super(getOpcode(CreateMode.fromFlag(flags, CreateMode.PERSISTENT)), path);
@@ -190,6 +242,9 @@ public abstract class Op {
         }
 
         private static int getOpcode(CreateMode createMode) {
+            if (createMode.isTTL()) {
+                return ZooDefs.OpCode.createTTL;
+            }
             return createMode.isContainer() ? ZooDefs.OpCode.createContainer : ZooDefs.OpCode.create;
         }
 
@@ -243,6 +298,48 @@ public abstract class Op {
         void validate() throws KeeperException {
             CreateMode createMode = CreateMode.fromFlag(flags);
             PathUtils.validatePath(getPath(), createMode.isSequential());
+            EphemeralType.validateTTL(createMode, -1);
+        }
+    }
+
+    public static class CreateTTL extends Create {
+        private final long ttl;
+
+        private CreateTTL(String path, byte[] data, List<ACL> acl, int flags, long ttl) {
+            super(path, data, acl, flags);
+            this.ttl = ttl;
+        }
+
+        private CreateTTL(String path, byte[] data, List<ACL> acl, CreateMode createMode, long ttl) {
+            super(path, data, acl, createMode);
+            this.ttl = ttl;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            return super.equals(o) && (o instanceof CreateTTL) && (ttl == ((CreateTTL)o).ttl);
+        }
+
+        @Override
+        public int hashCode() {
+            return super.hashCode() + (int)(ttl ^ (ttl >>> 32));
+        }
+
+        @Override
+        public Record toRequestRecord() {
+            return new CreateTTLRequest(getPath(), data, acl, flags, ttl);
+        }
+
+        @Override
+        Op withChroot(String path) {
+            return new CreateTTL(path, data, acl, flags, ttl);
+        }
+
+        @Override
+        void validate() throws KeeperException {
+            CreateMode createMode = CreateMode.fromFlag(flags);
+            PathUtils.validatePath(getPath(), createMode.isSequential());
+            EphemeralType.validateTTL(createMode, ttl);
         }
     }
 

src/java/main/org/apache/zookeeper/ZooDefs.java

@@ -69,6 +69,8 @@ public class ZooDefs {
 
         public final int deleteContainer = 20;
 
+        public final int createTTL = 21;
+
         public final int auth = 100;
 
         public final int setWatches = 101;

src/java/main/org/apache/zookeeper/ZooKeeper.java

@@ -18,18 +18,6 @@
 
 package org.apache.zookeeper;
 
-import java.io.IOException;
-import java.lang.reflect.Constructor;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
 import org.apache.jute.Record;
 import org.apache.zookeeper.AsyncCallback.ACLCallback;
 import org.apache.zookeeper.AsyncCallback.Children2Callback;
@@ -45,10 +33,10 @@ import org.apache.zookeeper.KeeperException.NoWatcherException;
 import org.apache.zookeeper.OpResult.ErrorResult;
 import org.apache.zookeeper.Watcher.Event.EventType;
 import org.apache.zookeeper.Watcher.WatcherType;
-import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.client.ConnectStringParser;
 import org.apache.zookeeper.client.HostProvider;
 import org.apache.zookeeper.client.StaticHostProvider;
+import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.client.ZooKeeperSaslClient;
 import org.apache.zookeeper.common.PathUtils;
 import org.apache.zookeeper.common.StringUtils;
@@ -58,6 +46,7 @@ import org.apache.zookeeper.proto.CheckWatchesRequest;
 import org.apache.zookeeper.proto.Create2Response;
 import org.apache.zookeeper.proto.CreateRequest;
 import org.apache.zookeeper.proto.CreateResponse;
+import org.apache.zookeeper.proto.CreateTTLRequest;
 import org.apache.zookeeper.proto.DeleteRequest;
 import org.apache.zookeeper.proto.ExistsRequest;
 import org.apache.zookeeper.proto.GetACLRequest;
@@ -79,9 +68,22 @@ import org.apache.zookeeper.proto.SetDataResponse;
 import org.apache.zookeeper.proto.SyncRequest;
 import org.apache.zookeeper.proto.SyncResponse;
 import org.apache.zookeeper.server.DataTree;
+import org.apache.zookeeper.server.EphemeralType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
 /**
  * This is the main class of ZooKeeper client library. To use a ZooKeeper
  * service, an application must first instantiate an object of ZooKeeper class.
@@ -1393,6 +1395,7 @@ public class ZooKeeper {
     {
         final String clientPath = path;
         PathUtils.validatePath(clientPath, createMode.isSequential());
+        EphemeralType.validateTTL(createMode, -1);
 
         final String serverPath = prependChroot(clientPath);
 
@@ -1479,23 +1482,34 @@ public class ZooKeeper {
     public String create(final String path, byte data[], List<ACL> acl,
             CreateMode createMode, Stat stat)
             throws KeeperException, InterruptedException {
+        return create(path, data, acl, createMode, stat, -1);
+    }
+
+    /**
+     * same as {@link #create(String, byte[], List, CreateMode, Stat)} but
+     * allows for specifying a TTL when mode is {@link CreateMode#PERSISTENT_WITH_TTL}
+     * or {@link CreateMode#PERSISTENT_SEQUENTIAL_WITH_TTL}. If the znode has not been modified
+     * within the given TTL, it will be deleted once it has no children. The TTL unit is
+     * milliseconds and must be greater than 0 and less than or equal to
+     * {@link EphemeralType#MAX_TTL}.
+     */
+    public String create(final String path, byte data[], List<ACL> acl,
+            CreateMode createMode, Stat stat, long ttl)
+            throws KeeperException, InterruptedException {
         final String clientPath = path;
         PathUtils.validatePath(clientPath, createMode.isSequential());
+        EphemeralType.validateTTL(createMode, ttl);
 
         final String serverPath = prependChroot(clientPath);
 
         RequestHeader h = new RequestHeader();
-        h.setType(createMode.isContainer() ? ZooDefs.OpCode.createContainer : ZooDefs.OpCode.create2);
-        CreateRequest request = new CreateRequest();
+        setCreateHeader(createMode, h);
         Create2Response response = new Create2Response();
-        request.setData(data);
-        request.setFlags(createMode.toFlag());
-        request.setPath(serverPath);
         if (acl != null && acl.size() == 0) {
             throw new KeeperException.InvalidACLException();
         }
-        request.setAcl(acl);
-        ReplyHeader r = cnxn.submitRequest(h, request, response, null);
+        Record record = makeCreateRecord(createMode, serverPath, data, acl, ttl);
+        ReplyHeader r = cnxn.submitRequest(h, record, response, null);
         if (r.getErr() != 0) {
             throw KeeperException.create(KeeperException.Code.get(r.getErr()),
                     clientPath);
@@ -1510,6 +1524,35 @@ public class ZooKeeper {
         }
     }
 
+    private void setCreateHeader(CreateMode createMode, RequestHeader h) {
+        if (createMode.isTTL()) {
+            h.setType(ZooDefs.OpCode.createTTL);
+        } else {
+            h.setType(createMode.isContainer() ? ZooDefs.OpCode.createContainer : ZooDefs.OpCode.create2);
+        }
+    }
+
+    private Record makeCreateRecord(CreateMode createMode, String serverPath, byte[] data, List<ACL> acl, long ttl) {
+        Record record;
+        if (createMode.isTTL()) {
+            CreateTTLRequest request = new CreateTTLRequest();
+            request.setData(data);
+            request.setFlags(createMode.toFlag());
+            request.setPath(serverPath);
+            request.setAcl(acl);
+            request.setTtl(ttl);
+            record = request;
+        } else {
+            CreateRequest request = new CreateRequest();
+            request.setData(data);
+            request.setFlags(createMode.toFlag());
+            request.setPath(serverPath);
+            request.setAcl(acl);
+            record = request;
+        }
+        return record;
+    }
+
     /**
      * The asynchronous version of create.
      *
@@ -1520,6 +1563,7 @@ public class ZooKeeper {
     {
         final String clientPath = path;
         PathUtils.validatePath(clientPath, createMode.isSequential());
+        EphemeralType.validateTTL(createMode, -1);
 
         final String serverPath = prependChroot(clientPath);
 
@@ -1543,22 +1587,30 @@ public class ZooKeeper {
      */
     public void create(final String path, byte data[], List<ACL> acl,
             CreateMode createMode, Create2Callback cb, Object ctx)
+    {
+        create(path, data, acl, createMode, cb, ctx, -1);
+    }
+
+    /**
+     * The asynchronous version of create with ttl.
+     *
+     * @see #create(String, byte[], List, CreateMode, Stat, long)
+     */
+    public void create(final String path, byte data[], List<ACL> acl,
+            CreateMode createMode, Create2Callback cb, Object ctx, long ttl)
     {
         final String clientPath = path;
         PathUtils.validatePath(clientPath, createMode.isSequential());
+        EphemeralType.validateTTL(createMode, ttl);
 
         final String serverPath = prependChroot(clientPath);
 
         RequestHeader h = new RequestHeader();
-        h.setType(createMode.isContainer() ? ZooDefs.OpCode.createContainer : ZooDefs.OpCode.create2);
-        CreateRequest request = new CreateRequest();
-        Create2Response response = new Create2Response();
+        setCreateHeader(createMode, h);
         ReplyHeader r = new ReplyHeader();
-        request.setData(data);
-        request.setFlags(createMode.toFlag());
-        request.setPath(serverPath);
-        request.setAcl(acl);
-        cnxn.queuePacket(h, r, request, response, cb, clientPath,
+        Create2Response response = new Create2Response();
+        Record record = makeCreateRecord(createMode, serverPath, data, acl, ttl);
+        cnxn.queuePacket(h, r, record, response, cb, clientPath,
                 serverPath, ctx, null);
     }
 

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

@@ -153,6 +153,24 @@ public class ContainerManager {
                 candidates.add(containerPath);
             }
         }
+        for (String ttlPath : zkDb.getDataTree().getTtls()) {
+            DataNode node = zkDb.getDataTree().getNode(ttlPath);
+            if (node != null) {
+                Set<String> children = node.getChildren();
+                if ((children == null) || (children.size() == 0)) {
+                    long elapsed = getElapsed(node);
+                    long ttl = EphemeralType.getTTL(node.stat.getEphemeralOwner());
+                    if ((ttl != 0) && (elapsed > ttl)) {
+                        candidates.add(ttlPath);
+                    }
+                }
+            }
+        }
         return candidates;
     }
+
+    // VisibleForTesting
+    protected long getElapsed(DataNode node) {
+        return Time.currentWallTime() - node.stat.getMtime();
+    }
 }

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

@@ -159,8 +159,11 @@ public class DataNode implements Record {
     }
 
     private static long getClientEphemeralOwner(StatPersisted stat) {
-        return (stat.getEphemeralOwner() == DataTree.CONTAINER_EPHEMERAL_OWNER)
-                ? 0 : stat.getEphemeralOwner();
+        EphemeralType ephemeralType = EphemeralType.get(stat.getEphemeralOwner());
+        if (ephemeralType != EphemeralType.NORMAL) {
+            return 0;
+        }
+        return stat.getEphemeralOwner();
     }
 
     synchronized public void deserialize(InputArchive archive, String tag)

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

@@ -41,6 +41,7 @@ import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.data.StatPersisted;
 import org.apache.zookeeper.txn.CheckVersionTxn;
 import org.apache.zookeeper.txn.CreateContainerTxn;
+import org.apache.zookeeper.txn.CreateTTLTxn;
 import org.apache.zookeeper.txn.CreateTxn;
 import org.apache.zookeeper.txn.DeleteTxn;
 import org.apache.zookeeper.txn.ErrorTxn;
@@ -78,8 +79,6 @@ import java.util.concurrent.ConcurrentHashMap;
 public class DataTree {
     private static final Logger LOG = LoggerFactory.getLogger(DataTree.class);
 
-    public static final long CONTAINER_EPHEMERAL_OWNER = Long.MIN_VALUE;
-
     /**
      * This hashtable provides a fast lookup to the datanodes. The tree is the
      * source of truth and is where all the locking occurs
@@ -137,6 +136,12 @@ public class DataTree {
     private final Set<String> containers =
             Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
 
+    /**
+     * This set contains the paths of all ttl nodes
+     */
+    private final Set<String> ttls =
+            Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+
     private final ReferenceCountedACLCache aclCache = new ReferenceCountedACLCache();
 
     @SuppressWarnings("unchecked")
@@ -156,6 +161,10 @@ public class DataTree {
         return new HashSet<String>(containers);
     }
 
+    public Set<String> getTtls() {
+        return new HashSet<String>(ttls);
+    }
+
     public Collection<Long> getSessions() {
         return ephemerals.keySet();
     }
@@ -455,8 +464,11 @@ public class DataTree {
             DataNode child = new DataNode(data, longval, stat);
             parent.addChild(childName);
             nodes.put(path, child);
-            if (ephemeralOwner == CONTAINER_EPHEMERAL_OWNER) {
+            EphemeralType ephemeralType = EphemeralType.get(ephemeralOwner);
+            if (ephemeralType == EphemeralType.CONTAINER) {
                 containers.add(path);
+            } else if (ephemeralType == EphemeralType.TTL) {
+                ttls.add(path);
             } else if (ephemeralOwner != 0) {
                 HashSet<String> list = ephemerals.get(ephemeralOwner);
                 if (list == null) {
@@ -526,8 +538,11 @@ public class DataTree {
             parent.removeChild(childName);
             parent.stat.setPzxid(zxid);
             long eowner = node.stat.getEphemeralOwner();
-            if (eowner == CONTAINER_EPHEMERAL_OWNER) {
+            EphemeralType ephemeralType = EphemeralType.get(eowner);
+            if (ephemeralType == EphemeralType.CONTAINER) {
                 containers.remove(path);
+            } else if (ephemeralType == EphemeralType.TTL) {
+                ttls.remove(path);
             } else if (eowner != 0) {
                 HashSet<String> nodes = ephemerals.get(eowner);
                 if (nodes != null) {
@@ -790,6 +805,19 @@ public class DataTree {
                             header.getZxid(), header.getTime(), stat);
                     rc.stat = stat;
                     break;
+                case OpCode.createTTL:
+                    CreateTTLTxn createTtlTxn = (CreateTTLTxn) txn;
+                    rc.path = createTtlTxn.getPath();
+                    stat = new Stat();
+                    createNode(
+                            createTtlTxn.getPath(),
+                            createTtlTxn.getData(),
+                            createTtlTxn.getAcl(),
+                            EphemeralType.ttlToEphemeralOwner(createTtlTxn.getTtl()),
+                            createTtlTxn.getParentCVersion(),
+                            header.getZxid(), header.getTime(), stat);
+                    rc.stat = stat;
+                    break;
                 case OpCode.createContainer:
                     CreateContainerTxn createContainerTxn = (CreateContainerTxn) txn;
                     rc.path = createContainerTxn.getPath();
@@ -798,7 +826,7 @@ public class DataTree {
                             createContainerTxn.getPath(),
                             createContainerTxn.getData(),
                             createContainerTxn.getAcl(),
-                            CONTAINER_EPHEMERAL_OWNER,
+                            EphemeralType.CONTAINER_EPHEMERAL_OWNER,
                             createContainerTxn.getParentCVersion(),
                             header.getZxid(), header.getTime(), stat);
                     rc.stat = stat;
@@ -854,6 +882,9 @@ public class DataTree {
                             case OpCode.create:
                                 record = new CreateTxn();
                                 break;
+                            case OpCode.createTTL:
+                                record = new CreateTTLTxn();
+                                break;
                             case OpCode.createContainer:
                                 record = new CreateContainerTxn();
                                 break;
@@ -1184,8 +1215,11 @@ public class DataTree {
                 }
                 parent.addChild(path.substring(lastSlash + 1));
                 long eowner = node.stat.getEphemeralOwner();
-                if (eowner == CONTAINER_EPHEMERAL_OWNER) {
+                EphemeralType ephemeralType = EphemeralType.get(eowner);
+                if (ephemeralType == EphemeralType.CONTAINER) {
                     containers.add(path);
+                } else if (ephemeralType == EphemeralType.TTL) {
+                    ttls.add(path);
                 } else if (eowner != 0) {
                     HashSet<String> list = ephemerals.get(eowner);
                     if (list == null) {

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

@@ -0,0 +1,77 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.server;
+
+import org.apache.zookeeper.CreateMode;
+
+
+public enum EphemeralType {
+    /**
+     * Not ephemeral
+     */
+    VOID,
+    /**
+     * Standard, pre-3.5.x EPHEMERAL
+     */
+    NORMAL,
+    /**
+     * Container node
+     */
+    CONTAINER,
+    /**
+     * TTL node
+     */
+    TTL;
+
+    public static final long CONTAINER_EPHEMERAL_OWNER = Long.MIN_VALUE;
+    public static final long MAX_TTL = 0x0fffffffffffffffL;
+    public static final long TTL_MASK = 0x8000000000000000L;
+
+    public static EphemeralType get(long ephemeralOwner) {
+        if (ephemeralOwner == CONTAINER_EPHEMERAL_OWNER) {
+            return CONTAINER;
+        }
+        if (ephemeralOwner < 0) {
+            return TTL;
+        }
+        return (ephemeralOwner == 0) ? VOID : NORMAL;
+    }
+
+    public static void validateTTL(CreateMode mode, long ttl) {
+        if (mode.isTTL()) {
+            ttlToEphemeralOwner(ttl);
+        } else if (ttl >= 0) {
+            throw new IllegalArgumentException("ttl not valid for mode: " + mode);
+        }
+    }
+
+    public static long getTTL(long ephemeralOwner) {
+        if ((ephemeralOwner < 0) && (ephemeralOwner != CONTAINER_EPHEMERAL_OWNER)) {
+            return (ephemeralOwner & MAX_TTL);
+        }
+        return 0;
+    }
+
+    public static long ttlToEphemeralOwner(long ttl) {
+        if ((ttl > MAX_TTL) || (ttl <= 0)) {
+            throw new IllegalArgumentException("ttl must be positive and cannot be larger than: " + MAX_TTL);
+        }
+        return TTL_MASK | ttl;
+    }
+}

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

@@ -215,6 +215,7 @@ public class FinalRequestProcessor implements RequestProcessor {
                             subResult = new CreateResult(subTxnResult.path);
                             break;
                         case OpCode.create2:
+                        case OpCode.createTTL:
                         case OpCode.createContainer:
                             subResult = new CreateResult(subTxnResult.path, subTxnResult.stat);
                             break;
@@ -244,6 +245,7 @@ public class FinalRequestProcessor implements RequestProcessor {
                 break;
             }
             case OpCode.create2:
+            case OpCode.createTTL:
             case OpCode.createContainer: {
                 lastOp = "CREA";
                 rsp = new Create2Response(rc.path, rc.stat);

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

@@ -36,6 +36,7 @@ import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.data.StatPersisted;
 import org.apache.zookeeper.proto.CheckVersionRequest;
 import org.apache.zookeeper.proto.CreateRequest;
+import org.apache.zookeeper.proto.CreateTTLRequest;
 import org.apache.zookeeper.proto.DeleteRequest;
 import org.apache.zookeeper.proto.ReconfigRequest;
 import org.apache.zookeeper.proto.SetACLRequest;
@@ -53,6 +54,7 @@ import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.txn.CheckVersionTxn;
 import org.apache.zookeeper.txn.CreateContainerTxn;
 import org.apache.zookeeper.txn.CreateSessionTxn;
+import org.apache.zookeeper.txn.CreateTTLTxn;
 import org.apache.zookeeper.txn.CreateTxn;
 import org.apache.zookeeper.txn.DeleteTxn;
 import org.apache.zookeeper.txn.ErrorTxn;
@@ -364,53 +366,9 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
         switch (type) {
             case OpCode.create:
             case OpCode.create2:
+            case OpCode.createTTL:
             case OpCode.createContainer: {
-                CreateRequest createRequest = (CreateRequest)record;
-                if (deserialize) {
-                    ByteBufferInputStream.byteBuffer2Record(request.request, createRequest);
-                }
-                CreateMode createMode = CreateMode.fromFlag(createRequest.getFlags());
-                validateCreateRequest(createMode, request);
-                String path = createRequest.getPath();
-                String parentPath = validatePathForCreate(path, request.sessionId);
-
-                List<ACL> listACL = fixupACL(path, request.authInfo, createRequest.getAcl());
-                ChangeRecord parentRecord = getRecordForPath(parentPath);
-
-                checkACL(zks, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo);
-                int parentCVersion = parentRecord.stat.getCversion();
-                if (createMode.isSequential()) {
-                    path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);
-                }
-                validatePath(path, request.sessionId);
-                try {
-                    if (getRecordForPath(path) != null) {
-                        throw new KeeperException.NodeExistsException(path);
-                    }
-                } catch (KeeperException.NoNodeException e) {
-                    // ignore this one
-                }
-                boolean ephemeralParent = (parentRecord.stat.getEphemeralOwner() != 0) &&
-                        (parentRecord.stat.getEphemeralOwner() != DataTree.CONTAINER_EPHEMERAL_OWNER);
-                if (ephemeralParent) {
-                    throw new KeeperException.NoChildrenForEphemeralsException(path);
-                }
-                int newCversion = parentRecord.stat.getCversion()+1;
-                if (type == OpCode.createContainer) {
-                    request.setTxn(new CreateContainerTxn(path, createRequest.getData(), listACL, newCversion));
-                } else {
-                    request.setTxn(new CreateTxn(path, createRequest.getData(), listACL, createMode.isEphemeral(),
-                            newCversion));
-                }
-                StatPersisted s = new StatPersisted();
-                if (createMode.isEphemeral()) {
-                    s.setEphemeralOwner(request.sessionId);
-                }
-                parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
-                parentRecord.childCount++;
-                parentRecord.stat.setCversion(newCversion);
-                addChangeRecord(parentRecord);
-                addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, s, 0, listACL));
+                pRequest2TxnCreate(type, request, record, deserialize);
                 break;
             }
             case OpCode.deleteContainer: {
@@ -421,7 +379,7 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 if (nodeRecord.childCount > 0) {
                     throw new KeeperException.NotEmptyException(path);
                 }
-                if (nodeRecord.stat.getEphemeralOwner() != DataTree.CONTAINER_EPHEMERAL_OWNER) {
+                if (EphemeralType.get(nodeRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL) {
                     throw new KeeperException.BadVersionException(path);
                 }
                 request.setTxn(new DeleteTxn(path));
@@ -665,6 +623,75 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
         }
     }
 
+    private void pRequest2TxnCreate(int type, Request request, Record record, boolean deserialize) throws IOException, KeeperException {
+        if (deserialize) {
+            ByteBufferInputStream.byteBuffer2Record(request.request, record);
+        }
+
+        int flags;
+        String path;
+        List<ACL> acl;
+        byte[] data;
+        long ttl;
+        if (type == OpCode.createTTL) {
+            CreateTTLRequest createTtlRequest = (CreateTTLRequest)record;
+            flags = createTtlRequest.getFlags();
+            path = createTtlRequest.getPath();
+            acl = createTtlRequest.getAcl();
+            data = createTtlRequest.getData();
+            ttl = createTtlRequest.getTtl();
+        } else {
+            CreateRequest createRequest = (CreateRequest)record;
+            flags = createRequest.getFlags();
+            path = createRequest.getPath();
+            acl = createRequest.getAcl();
+            data = createRequest.getData();
+            ttl = 0;
+        }
+        CreateMode createMode = CreateMode.fromFlag(flags);
+        validateCreateRequest(createMode, request);
+        String parentPath = validatePathForCreate(path, request.sessionId);
+
+        List<ACL> listACL = fixupACL(path, request.authInfo, acl);
+        ChangeRecord parentRecord = getRecordForPath(parentPath);
+
+        checkACL(zks, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo);
+        int parentCVersion = parentRecord.stat.getCversion();
+        if (createMode.isSequential()) {
+            path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);
+        }
+        validatePath(path, request.sessionId);
+        try {
+            if (getRecordForPath(path) != null) {
+                throw new KeeperException.NodeExistsException(path);
+            }
+        } catch (KeeperException.NoNodeException e) {
+            // ignore this one
+        }
+        boolean ephemeralParent = EphemeralType.get(parentRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL;
+        if (ephemeralParent) {
+            throw new KeeperException.NoChildrenForEphemeralsException(path);
+        }
+        int newCversion = parentRecord.stat.getCversion()+1;
+        if (type == OpCode.createContainer) {
+            request.setTxn(new CreateContainerTxn(path, data, listACL, newCversion));
+        } else if (type == OpCode.createTTL) {
+            request.setTxn(new CreateTTLTxn(path, data, listACL, newCversion, ttl));
+        } else {
+            request.setTxn(new CreateTxn(path, data, listACL, createMode.isEphemeral(),
+                    newCversion));
+        }
+        StatPersisted s = new StatPersisted();
+        if (createMode.isEphemeral()) {
+            s.setEphemeralOwner(request.sessionId);
+        }
+        parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
+        parentRecord.childCount++;
+        parentRecord.stat.setCversion(newCversion);
+        addChangeRecord(parentRecord);
+        addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, s, 0, listACL));
+    }
+
     private void validatePath(String path, long sessionId) throws BadArgumentsException {
         try {
             PathUtils.validatePath(path);
@@ -713,6 +740,10 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 CreateRequest create2Request = new CreateRequest();
                 pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true);
                 break;
+            case OpCode.createTTL:
+                CreateTTLRequest createTtlRequest = new CreateTTLRequest();
+                pRequest2Txn(request.type, zks.getNextZxid(), request, createTtlRequest, true);
+                break;
             case OpCode.deleteContainer:
             case OpCode.delete:
                 DeleteRequest deleteRequest = new DeleteRequest();

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

@@ -137,6 +137,7 @@ public class Request {
         case OpCode.closeSession:
         case OpCode.create:
         case OpCode.create2:
+        case OpCode.createTTL:
         case OpCode.createContainer:
         case OpCode.createSession:
         case OpCode.delete:
@@ -171,6 +172,7 @@ public class Request {
             return false;
         case OpCode.create:
         case OpCode.create2:
+        case OpCode.createTTL:
         case OpCode.createContainer:
         case OpCode.error:
         case OpCode.delete:
@@ -197,6 +199,8 @@ public class Request {
             return "create";
         case OpCode.create2:
             return "create2";
+        case OpCode.createTTL:
+            return "createTtl";
         case OpCode.createContainer:
             return "createContainer";
         case OpCode.setWatches:

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

@@ -37,6 +37,8 @@ public class TraceFormatter {
             return "create";
         case OpCode.create2:
             return "create2";
+        case OpCode.createTTL:
+            return "createTtl";
         case OpCode.createContainer:
             return "createContainer";
         case OpCode.delete:

src/java/main/org/apache/zookeeper/server/quorum/CommitProcessor.java

@@ -133,6 +133,7 @@ public class CommitProcessor extends ZooKeeperCriticalThread implements
         switch (request.type) {
             case OpCode.create:
             case OpCode.create2:
+            case OpCode.createTTL:
             case OpCode.createContainer:
             case OpCode.delete:
             case OpCode.deleteContainer:

src/java/main/org/apache/zookeeper/server/quorum/FollowerRequestProcessor.java

@@ -84,6 +84,7 @@ public class FollowerRequestProcessor extends ZooKeeperCriticalThread implements
                     break;
                 case OpCode.create:
                 case OpCode.create2:
+                case OpCode.createTTL:
                 case OpCode.createContainer:
                 case OpCode.delete:
                 case OpCode.deleteContainer:

src/java/main/org/apache/zookeeper/server/quorum/ObserverRequestProcessor.java

@@ -93,6 +93,7 @@ public class ObserverRequestProcessor extends ZooKeeperCriticalThread implements
                     break;
                 case OpCode.create:
                 case OpCode.create2:
+                case OpCode.createTTL:
                 case OpCode.createContainer:
                 case OpCode.delete:
                 case OpCode.deleteContainer:

src/java/main/org/apache/zookeeper/server/quorum/ReadOnlyRequestProcessor.java

@@ -82,6 +82,7 @@ public class ReadOnlyRequestProcessor extends ZooKeeperCriticalThread implements
                 case OpCode.sync:
                 case OpCode.create:
                 case OpCode.create2:
+                case OpCode.createTTL:
                 case OpCode.createContainer:
                 case OpCode.delete:
                 case OpCode.deleteContainer:

src/java/main/org/apache/zookeeper/server/util/SerializeUtils.java

@@ -27,6 +27,7 @@ import org.apache.zookeeper.server.DataTree;
 import org.apache.zookeeper.server.ZooTrace;
 import org.apache.zookeeper.txn.CreateContainerTxn;
 import org.apache.zookeeper.txn.CreateSessionTxn;
+import org.apache.zookeeper.txn.CreateTTLTxn;
 import org.apache.zookeeper.txn.CreateTxn;
 import org.apache.zookeeper.txn.CreateTxnV0;
 import org.apache.zookeeper.txn.DeleteTxn;
@@ -68,6 +69,9 @@ public class SerializeUtils {
         case OpCode.create2:
             txn = new CreateTxn();
             break;
+        case OpCode.createTTL:
+            txn = new CreateTTLTxn();
+            break;
         case OpCode.createContainer:
             txn = new CreateContainerTxn();
             break;

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

@@ -0,0 +1,213 @@
+/**
+ * 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.server;
+
+import org.apache.zookeeper.AsyncCallback;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.OpResult;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.test.ClientBase;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class CreateTTLTest extends ClientBase {
+    private ZooKeeper zk;
+
+    @Override
+    public void setUp() throws Exception {
+        super.setUp();
+        zk = createClient();
+    }
+
+    @Override
+    public void tearDown() throws Exception {
+        super.tearDown();
+        zk.close();
+    }
+
+    @Test
+    public void testCreate()
+            throws IOException, KeeperException, InterruptedException {
+        Stat stat = new Stat();
+        zk.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_WITH_TTL, stat, 100);
+        Assert.assertEquals(0, stat.getEphemeralOwner());
+
+        final AtomicLong fakeElapsed = new AtomicLong(0);
+        ContainerManager containerManager = newContainerManager(fakeElapsed);
+        containerManager.checkContainers();
+        Assert.assertNotNull("Ttl node should not have been deleted yet", zk.exists("/foo", false));
+
+        fakeElapsed.set(1000);
+        containerManager.checkContainers();
+        Assert.assertNull("Ttl node should have been deleted", zk.exists("/foo", false));
+    }
+
+    @Test
+    public void testCreateSequential()
+            throws IOException, KeeperException, InterruptedException {
+        Stat stat = new Stat();
+        String path = zk.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL, stat, 100);
+        Assert.assertEquals(0, stat.getEphemeralOwner());
+
+        final AtomicLong fakeElapsed = new AtomicLong(0);
+        ContainerManager containerManager = newContainerManager(fakeElapsed);
+        containerManager.checkContainers();
+        Assert.assertNotNull("Ttl node should not have been deleted yet", zk.exists(path, false));
+
+        fakeElapsed.set(1000);
+        containerManager.checkContainers();
+        Assert.assertNull("Ttl node should have been deleted", zk.exists(path, false));
+    }
+
+    @Test
+    public void testCreateAsync()
+            throws IOException, KeeperException, InterruptedException {
+        AsyncCallback.Create2Callback callback = new AsyncCallback.Create2Callback() {
+            @Override
+            public void processResult(int rc, String path, Object ctx, String name, Stat stat) {
+                // NOP
+            }
+        };
+        zk.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_WITH_TTL, callback, null, 100);
+
+        final AtomicLong fakeElapsed = new AtomicLong(0);
+        ContainerManager containerManager = newContainerManager(fakeElapsed);
+        containerManager.checkContainers();
+        Assert.assertNotNull("Ttl node should not have been deleted yet", zk.exists("/foo", false));
+
+        fakeElapsed.set(1000);
+        containerManager.checkContainers();
+        Assert.assertNull("Ttl node should have been deleted", zk.exists("/foo", false));
+    }
+
+    @Test
+    public void testModifying()
+            throws IOException, KeeperException, InterruptedException {
+        Stat stat = new Stat();
+        zk.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_WITH_TTL, stat, 100);
+        Assert.assertEquals(0, stat.getEphemeralOwner());
+
+        final AtomicLong fakeElapsed = new AtomicLong(0);
+        ContainerManager containerManager = newContainerManager(fakeElapsed);
+        containerManager.checkContainers();
+        Assert.assertNotNull("Ttl node should not have been deleted yet", zk.exists("/foo", false));
+
+        for ( int i = 0; i < 10; ++i ) {
+            fakeElapsed.set(50);
+            zk.setData("/foo", new byte[i + 1], -1);
+            containerManager.checkContainers();
+            Assert.assertNotNull("Ttl node should not have been deleted yet", zk.exists("/foo", false));
+        }
+
+        fakeElapsed.set(200);
+        containerManager.checkContainers();
+        Assert.assertNull("Ttl node should have been deleted", zk.exists("/foo", false));
+    }
+
+    @Test
+    public void testMulti()
+            throws IOException, 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);
+        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();
+
+        final AtomicLong fakeElapsed = new AtomicLong(0);
+        ContainerManager containerManager = newContainerManager(fakeElapsed);
+        containerManager.checkContainers();
+        Assert.assertNotNull("node should not have been deleted yet", zk.exists("/a", false));
+        Assert.assertNotNull("node should not have been deleted yet", zk.exists(sequentialPath, false));
+        Assert.assertNotNull("node should never be deleted", zk.exists("/c", false));
+
+        fakeElapsed.set(110);
+        containerManager.checkContainers();
+        Assert.assertNull("node should have been deleted", zk.exists("/a", false));
+        Assert.assertNotNull("node should not have been deleted yet", zk.exists(sequentialPath, false));
+        Assert.assertNotNull("node should never be deleted", zk.exists("/c", false));
+
+        fakeElapsed.set(210);
+        containerManager.checkContainers();
+        Assert.assertNull("node should have been deleted", zk.exists("/a", false));
+        Assert.assertNull("node should have been deleted", zk.exists(sequentialPath, false));
+        Assert.assertNotNull("node should never be deleted", zk.exists("/c", false));
+    }
+
+    @Test
+    public void testBadUsage() throws KeeperException, InterruptedException {
+        for ( CreateMode createMode : CreateMode.values() ) {
+            try {
+                zk.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode, new Stat(), createMode.isTTL() ? 0 : 100);
+                Assert.fail("should have thrown IllegalArgumentException");
+            } catch (IllegalArgumentException dummy) {
+                // correct
+            }
+        }
+
+        for ( CreateMode createMode : CreateMode.values() ) {
+            AsyncCallback.Create2Callback callback = new AsyncCallback.Create2Callback() {
+                @Override
+                public void processResult(int rc, String path, Object ctx, String name, Stat stat) {
+                    // NOP
+                }
+            };
+            try {
+                zk.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode, callback, null, createMode.isTTL() ? 0 : 100);
+                Assert.fail("should have thrown IllegalArgumentException");
+            } catch (IllegalArgumentException dummy) {
+                // correct
+            }
+        }
+
+        try {
+            Op op = Op.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_WITH_TTL, 0);
+            zk.multi(Collections.singleton(op));
+            Assert.fail("should have thrown IllegalArgumentException");
+        } catch (IllegalArgumentException dummy) {
+            // correct
+        }
+        try {
+            Op op = Op.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL, 0);
+            zk.multi(Collections.singleton(op));
+            Assert.fail("should have thrown IllegalArgumentException");
+        } catch (IllegalArgumentException dummy) {
+            // correct
+        }
+    }
+
+    private ContainerManager newContainerManager(final AtomicLong fakeElapsed) {
+        return new ContainerManager(serverFactory.getZooKeeperServer()
+                .getZKDatabase(), serverFactory.getZooKeeperServer().firstProcessor, 1, 100) {
+            @Override
+            protected long getElapsed(DataNode node) {
+                return fakeElapsed.get();
+            }
+        };
+    }
+}

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

@@ -0,0 +1,58 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.server;
+
+import org.apache.zookeeper.CreateMode;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class EphemeralTypeTest {
+    @Test
+    public void testTtls() {
+        long ttls[] = {100, 1, EphemeralType.MAX_TTL};
+        for (long ttl : ttls) {
+            long ephemeralOwner = EphemeralType.ttlToEphemeralOwner(ttl);
+            Assert.assertEquals(EphemeralType.TTL, EphemeralType.get(ephemeralOwner));
+            Assert.assertEquals(ttl, EphemeralType.getTTL(ephemeralOwner));
+        }
+
+        EphemeralType.validateTTL(CreateMode.PERSISTENT_WITH_TTL, 100);
+        EphemeralType.validateTTL(CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL, 100);
+
+        try {
+            EphemeralType.validateTTL(CreateMode.EPHEMERAL, 100);
+            Assert.fail("Should have thrown IllegalArgumentException");
+        } catch (IllegalArgumentException dummy) {
+            // expected
+        }
+    }
+
+    @Test
+    public void testContainerValue() {
+        Assert.assertEquals(Long.MIN_VALUE, EphemeralType.CONTAINER_EPHEMERAL_OWNER);
+        Assert.assertEquals(EphemeralType.CONTAINER, EphemeralType.get(EphemeralType.CONTAINER_EPHEMERAL_OWNER));
+    }
+
+    @Test
+    public void testNonSpecial() {
+        Assert.assertEquals(EphemeralType.VOID, EphemeralType.get(0));
+        Assert.assertEquals(EphemeralType.NORMAL, EphemeralType.get(1));
+        Assert.assertEquals(EphemeralType.NORMAL, EphemeralType.get(Long.MAX_VALUE));
+    }
+}

src/zookeeper.jute

@@ -127,6 +127,13 @@ module org.apache.zookeeper.proto {
         vector<org.apache.zookeeper.data.ACL> acl;
         int flags;
     }
+    class CreateTTLRequest {
+        ustring path;
+        buffer data;
+        vector<org.apache.zookeeper.data.ACL> acl;
+        int flags;
+        long ttl;
+    }
     class DeleteRequest {
         ustring path;
         int version;
@@ -260,6 +267,13 @@ module org.apache.zookeeper.txn {
         boolean ephemeral;
         int parentCVersion;
     }
+    class CreateTTLTxn {
+        ustring path;
+        buffer data;
+        vector<org.apache.zookeeper.data.ACL> acl;
+        int parentCVersion;
+        long ttl;
+    }
     class CreateContainerTxn {
         ustring path;
         buffer data;