ZOOKEEPER-2609: Add TTL Node APIs to C client

* added zoo_acreate_ttl, zoo_acreate2_ttl, zoo_create_ttl,
  zoo_create2_ttl methods, where ttl (int64_t) can be defined.
* renamed flags to mode in zoo_*create* because it is not a bitmask on the
  server side.
* added Create mode contants to match CreateMode.java enum fields.
  ZOO_SEQUENCE is kept for backward compatibility.
  ZOO_EPHEMERAL|ZOO_SEQUENCE is still usable, but ZOO_EPHEMERAL_SEQUENTIAL
  is preferred.
* added / fixed doxygen comments for create methods.
* modified tests/zkServer.sh to pass system properties to ZK.
* added +t, +c shell commands to zookeeper cli.
* added a test case for create ttl.

Author: Balazs Meszaros <balazs.meszaros@cloudera.com>

Reviewers: andor@apache.org

Closes #891 from meszibalu/my/ttl

zookeeper-client/zookeeper-client-c/include/proto.h

@@ -42,6 +42,7 @@ extern "C" {
 #define ZOO_REMOVE_WATCHES 18
 #define ZOO_CREATE_CONTAINER_OP 19
 #define ZOO_DELETE_CONTAINER_OP 20
+#define ZOO_CREATE_TTL_OP 21
 #define ZOO_CLOSE_OP -11
 #define ZOO_SETAUTH_OP 100
 #define ZOO_SETWATCHES_OP 101

zookeeper-client/zookeeper-client-c/include/zookeeper.h

@@ -189,15 +189,24 @@ extern ZOOAPI const int ZOOKEEPER_READ;
 // @}
 
 /**
- * @name Create Flags
+ * @name Create Mode
  *
- * These flags are used by zoo_create to affect node create. They may
- * be ORed together to combine effects.
+ * These modes are used by zoo_create to affect node create.
  */
 // @{
+extern ZOOAPI const int ZOO_PERSISTENT;
 extern ZOOAPI const int ZOO_EPHEMERAL;
-extern ZOOAPI const int ZOO_SEQUENCE;
+extern ZOOAPI const int ZOO_PERSISTENT_SEQUENTIAL;
+extern ZOOAPI const int ZOO_EPHEMERAL_SEQUENTIAL;
 extern ZOOAPI const int ZOO_CONTAINER;
+extern ZOOAPI const int ZOO_PERSISTENT_WITH_TTL;
+extern ZOOAPI const int ZOO_PERSISTENT_SEQUENTIAL_WITH_TTL;
+
+/**
+ * \deprecated ZOO_SEQUENCE Create Flag has been deprecated. Use ZOO_PERSISTENT_SEQUENTIAL
+ * or ZOO_EPHEMERAL_SEQUENTIAL instead of it.
+ */
+extern ZOOAPI const int ZOO_SEQUENCE;
 // @}
 
 /**
@@ -307,6 +316,7 @@ typedef struct zoo_op {
             int buflen;
             const struct ACL_vector *acl;
             int flags;
+            int64_t ttl;
         } create_op;
 
         // DELETE
@@ -344,8 +354,7 @@ typedef struct zoo_op {
  * \param valuelen The number of bytes in data. To set the data to be NULL use
  * value as NULL and valuelen as -1.
  * \param acl The initial ACL of the node. The ACL must not be null or empty.
- * \param flags this parameter can be set to 0 for normal create or an OR
- *    of the Create Flags
+ * \param mode this parameter should be one of the Create Modes.
  * \param path_buffer Buffer which will be filled with the path of the
  *    new node (this might be different than the supplied path
  *    because of the ZOO_SEQUENCE flag).  The path string will always be
@@ -357,7 +366,7 @@ typedef struct zoo_op {
  *    The path string will always be null-terminated.
  */
 void zoo_create_op_init(zoo_op_t *op, const char *path, const char *value,
-        int valuelen,  const struct ACL_vector *acl, int flags,
+        int valuelen,  const struct ACL_vector *acl, int mode,
         char *path_buffer, int path_buffer_len);
 
 /**
@@ -896,11 +905,12 @@ ZOOAPI int zoo_state(zhandle_t *zh);
  * \brief create a node.
  *
  * This method will create a node in ZooKeeper. A node can only be created if
- * it does not already exists. The Create Flags affect the creation of nodes.
- * If ZOO_EPHEMERAL flag is set, the node will automatically get removed if the
- * client session goes away. If the ZOO_SEQUENCE flag is set, a unique
- * monotonically increasing sequence number is appended to the path name. The
- * sequence number is always fixed length of 10 digits, 0 padded.
+ * it does not already exist. The Create Mode affects the creation of nodes.
+ * If ZOO_EPHEMERAL mode is chosen, the node will automatically get removed if the
+ * client session goes away. If ZOO_CONTAINER flag is set, a container node will be
+ * created. For ZOO_*_SEQUENTIAL modes, a unique monotonically increasing
+ * sequence number is appended to the path name. The sequence number is always fixed
+ * length of 10 digits, 0 padded.
  *
  * \param zh the zookeeper handle obtained by a call to \ref zookeeper_init
  * \param path The name of the node. Expressed as a file name with slashes
@@ -908,8 +918,7 @@ ZOOAPI int zoo_state(zhandle_t *zh);
  * \param value The data to be stored in the node.
  * \param valuelen The number of bytes in data.
  * \param acl The initial ACL of the node. The ACL must not be null or empty.
- * \param flags this parameter can be set to 0 for normal create or an OR
- *    of the Create Flags
+ * \param mode this parameter should be one of the Create Modes.
  * \param completion the routine to invoke when the request completes. The completion
  * will be triggered with one of the following codes passed in as the rc argument:
  * ZOK operation completed successfully
@@ -925,18 +934,58 @@ ZOOAPI int zoo_state(zhandle_t *zh);
  * ZMARSHALLINGERROR - failed to marshall a request; possibly, out of memory
  */
 ZOOAPI int zoo_acreate(zhandle_t *zh, const char *path, const char *value, 
-        int valuelen, const struct ACL_vector *acl, int flags,
+        int valuelen, const struct ACL_vector *acl, int mode,
+        string_completion_t completion, const void *data);
+
+/**
+ * \brief create a node.
+ *
+ * This method will create a node in ZooKeeper. A node can only be created if
+ * it does not already exist. The Create Mode affects the creation of nodes.
+ * If ZOO_EPHEMERAL mode is chosen, the node will automatically get removed if the
+ * client session goes away. If ZOO_CONTAINER flag is set, a container node will be
+ * created. For ZOO_*_SEQUENTIAL modes, a unique monotonically increasing
+ * sequence number is appended to the path name. The sequence number is always fixed
+ * length of 10 digits, 0 padded. When ZOO_*_WITH_TTL is selected, a ttl node will be
+ * created.
+ *
+ * \param zh the zookeeper handle obtained by a call to \ref zookeeper_init
+ * \param path The name of the node. Expressed as a file name with slashes
+ * separating ancestors of the node.
+ * \param value The data to be stored in the node.
+ * \param valuelen The number of bytes in data.
+ * \param acl The initial ACL of the node. The ACL must not be null or empty.
+ * \param mode this parameter should be one of the Create Modes.
+ * \param ttl the value of ttl in milliseconds. It must be positive for ZOO_*_WITH_TTL
+ *    Create modes, otherwise it must be -1.
+ * \param completion the routine to invoke when the request completes. The completion
+ * will be triggered with one of the following codes passed in as the rc argument:
+ * ZOK operation completed successfully
+ * ZNONODE the parent node does not exist.
+ * ZNODEEXISTS the node already exists
+ * ZNOAUTH the client does not have permission.
+ * ZNOCHILDRENFOREPHEMERALS cannot create children of ephemeral nodes.
+ * \param data The data that will be passed to the completion routine when the
+ * function completes.
+ * \return ZOK on success or one of the following errcodes on failure:
+ * ZBADARGUMENTS - invalid input parameters
+ * ZINVALIDSTATE - zhandle state is either ZOO_SESSION_EXPIRED_STATE or ZOO_AUTH_FAILED_STATE
+ * ZMARSHALLINGERROR - failed to marshall a request; possibly, out of memory
+ */
+ZOOAPI int zoo_acreate_ttl(zhandle_t *zh, const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl, int mode, int64_t ttl,
         string_completion_t completion, const void *data);
 
 /**
  * \brief create a node asynchronously and returns stat details.
  *
  * This method will create a node in ZooKeeper. A node can only be created if
- * it does not already exists. The Create Flags affect the creation of nodes.
- * If ZOO_EPHEMERAL flag is set, the node will automatically get removed if the
- * client session goes away. If the ZOO_SEQUENCE flag is set, a unique
- * monotonically increasing sequence number is appended to the path name. The
- * sequence number is always fixed length of 10 digits, 0 padded.
+ * it does not already exist. The Create Mode affects the creation of nodes.
+ * If ZOO_EPHEMERAL mode is chosen, the node will automatically get removed if the
+ * client session goes away. If ZOO_CONTAINER flag is set, a container node will be
+ * created. For ZOO_*_SEQUENTIAL modes, a unique monotonically increasing
+ * sequence number is appended to the path name. The sequence number is always fixed
+ * length of 10 digits, 0 padded.
  *
  * \param zh the zookeeper handle obtained by a call to \ref zookeeper_init
  * \param path The name of the node. Expressed as a file name with slashes
@@ -944,8 +993,7 @@ ZOOAPI int zoo_acreate(zhandle_t *zh, const char *path, const char *value,
  * \param value The data to be stored in the node.
  * \param valuelen The number of bytes in data.
  * \param acl The initial ACL of the node. The ACL must not be null or empty.
- * \param flags this parameter can be set to 0 for normal create or an OR
- *    of the Create Flags
+ * \param mode this parameter should be one of the Create Modes.
  * \param completion the routine to invoke when the request completes. The completion
  * will be triggered with one of the following codes passed in as the rc argument:
  * ZOK operation completed successfully
@@ -961,7 +1009,46 @@ ZOOAPI int zoo_acreate(zhandle_t *zh, const char *path, const char *value,
  * ZMARSHALLINGERROR - failed to marshall a request; possibly, out of memory
  */
 ZOOAPI int zoo_acreate2(zhandle_t *zh, const char *path, const char *value,
-        int valuelen, const struct ACL_vector *acl, int flags,
+        int valuelen, const struct ACL_vector *acl, int mode,
+        string_stat_completion_t completion, const void *data);
+
+/**
+ * \brief create a node asynchronously and returns stat details.
+ *
+ * This method will create a node in ZooKeeper. A node can only be created if
+ * it does not already exist. The Create Mode affects the creation of nodes.
+ * If ZOO_EPHEMERAL mode is chosen, the node will automatically get removed if the
+ * client session goes away. If ZOO_CONTAINER flag is set, a container node will be
+ * created. For ZOO_*_SEQUENTIAL modes, a unique monotonically increasing
+ * sequence number is appended to the path name. The sequence number is always fixed
+ * length of 10 digits, 0 padded. When ZOO_*_WITH_TTL is selected, a ttl node will be
+ * created.
+ *
+ * \param zh the zookeeper handle obtained by a call to \ref zookeeper_init
+ * \param path The name of the node. Expressed as a file name with slashes
+ * separating ancestors of the node.
+ * \param value The data to be stored in the node.
+ * \param valuelen The number of bytes in data.
+ * \param acl The initial ACL of the node. The ACL must not be null or empty.
+ * \param mode this parameter should be one of the Create Modes.
+ * \param ttl the value of ttl in milliseconds. It must be positive for ZOO_*_WITH_TTL
+ *    Create modes, otherwise it must be -1.
+ * \param completion the routine to invoke when the request completes. The completion
+ * will be triggered with one of the following codes passed in as the rc argument:
+ * ZOK operation completed successfully
+ * ZNONODE the parent node does not exist.
+ * ZNODEEXISTS the node already exists
+ * ZNOAUTH the client does not have permission.
+ * ZNOCHILDRENFOREPHEMERALS cannot create children of ephemeral nodes.
+ * \param data The data that will be passed to the completion routine when the
+ * function completes.
+ * \return ZOK on success or one of the following errcodes on failure:
+ * ZBADARGUMENTS - invalid input parameters
+ * ZINVALIDSTATE - zhandle state is either ZOO_SESSION_EXPIRED_STATE or ZOO_AUTH_FAILED_STATE
+ * ZMARSHALLINGERROR - failed to marshall a request; possibly, out of memory
+ */
+ZOOAPI int zoo_acreate2_ttl(zhandle_t *zh, const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl, int mode, int64_t ttl,
         string_stat_completion_t completion, const void *data);
 
 /**
@@ -1593,10 +1680,12 @@ ZOOAPI int zoo_aremove_all_watches(zhandle_t *zh, const char *path,
  * \brief create a node synchronously.
  *
  * This method will create a node in ZooKeeper. A node can only be created if
- * it does not already exists. The Create Flags affect the creation of nodes.
- * If ZOO_EPHEMERAL flag is set, the node will automatically get removed if the
- * client session goes away. If the ZOO_SEQUENCE flag is set, a unique
- * monotonically increasing sequence number is appended to the path name.
+ * it does not already exist. The Create Mode affects the creation of nodes.
+ * If ZOO_EPHEMERAL mode is chosen, the node will automatically get removed if the
+ * client session goes away. If ZOO_CONTAINER flag is set, a container node will be
+ * created. For ZOO_*_SEQUENTIAL modes, a unique monotonically increasing
+ * sequence number is appended to the path name. The sequence number is always fixed
+ * length of 10 digits, 0 padded.
  *
  * \param zh the zookeeper handle obtained by a call to \ref zookeeper_init
  * \param path The name of the node. Expressed as a file name with slashes
@@ -1605,8 +1694,7 @@ ZOOAPI int zoo_aremove_all_watches(zhandle_t *zh, const char *path,
  * \param valuelen The number of bytes in data. To set the data to be NULL use
  * value as NULL and valuelen as -1.
  * \param acl The initial ACL of the node. The ACL must not be null or empty.
- * \param flags this parameter can be set to 0 for normal create or an OR
- *    of the Create Flags
+ * \param mode this parameter should be one of the Create Modes.
  * \param path_buffer Buffer which will be filled with the path of the
  *    new node (this might be different than the supplied path
  *    because of the ZOO_SEQUENCE flag).  The path string will always be
@@ -1627,17 +1715,64 @@ ZOOAPI int zoo_aremove_all_watches(zhandle_t *zh, const char *path,
  * ZMARSHALLINGERROR - failed to marshall a request; possibly, out of memory
  */
 ZOOAPI int zoo_create(zhandle_t *zh, const char *path, const char *value,
-        int valuelen, const struct ACL_vector *acl, int flags,
+        int valuelen, const struct ACL_vector *acl, int mode,
+        char *path_buffer, int path_buffer_len);
+
+/**
+ * \brief create a node synchronously.
+ *
+ * This method will create a node in ZooKeeper. A node can only be created if
+ * it does not already exist. The Create Mode affects the creation of nodes.
+ * If ZOO_EPHEMERAL mode is chosen, the node will automatically get removed if the
+ * client session goes away. If ZOO_CONTAINER flag is set, a container node will be
+ * created. For ZOO_*_SEQUENTIAL modes, a unique monotonically increasing
+ * sequence number is appended to the path name. The sequence number is always fixed
+ * length of 10 digits, 0 padded. When ZOO_*_WITH_TTL is selected, a ttl node will be
+ * created.
+ *
+ * \param zh the zookeeper handle obtained by a call to \ref zookeeper_init
+ * \param path The name of the node. Expressed as a file name with slashes
+ * separating ancestors of the node.
+ * \param value The data to be stored in the node.
+ * \param valuelen The number of bytes in data. To set the data to be NULL use
+ * value as NULL and valuelen as -1.
+ * \param acl The initial ACL of the node. The ACL must not be null or empty.
+ * \param mode this parameter should be one of the Create Modes.
+ * \param ttl the value of ttl in milliseconds. It must be positive for ZOO_*_WITH_TTL
+ *    Create modes, otherwise it must be -1.
+ * \param path_buffer Buffer which will be filled with the path of the
+ *    new node (this might be different than the supplied path
+ *    because of the ZOO_SEQUENCE flag).  The path string will always be
+ *    null-terminated. This parameter may be NULL if path_buffer_len = 0.
+ * \param path_buffer_len Size of path buffer; if the path of the new
+ *    node (including space for the null terminator) exceeds the buffer size,
+ *    the path string will be truncated to fit.  The actual path of the
+ *    new node in the server will not be affected by the truncation.
+ *    The path string will always be null-terminated.
+ * \return  one of the following codes are returned:
+ * ZOK operation completed successfully
+ * ZNONODE the parent node does not exist.
+ * ZNODEEXISTS the node already exists
+ * ZNOAUTH the client does not have permission.
+ * ZNOCHILDRENFOREPHEMERALS cannot create children of ephemeral nodes.
+ * ZBADARGUMENTS - invalid input parameters
+ * ZINVALIDSTATE - zhandle state is either ZOO_SESSION_EXPIRED_STATE or ZOO_AUTH_FAILED_STATE
+ * ZMARSHALLINGERROR - failed to marshall a request; possibly, out of memory
+ */
+ZOOAPI int zoo_create_ttl(zhandle_t *zh, const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl, int mode, int64_t ttl,
         char *path_buffer, int path_buffer_len);
 
 /**
  * \brief create a node synchronously and collect stat details.
  *
  * This method will create a node in ZooKeeper. A node can only be created if
- * it does not already exists. The Create Flags affect the creation of nodes.
- * If ZOO_EPHEMERAL flag is set, the node will automatically get removed if the
- * client session goes away. If the ZOO_SEQUENCE flag is set, a unique
- * monotonically increasing sequence number is appended to the path name.
+ * it does not already exist. The Create Mode affects the creation of nodes.
+ * If ZOO_EPHEMERAL mode is chosen, the node will automatically get removed if the
+ * client session goes away. If ZOO_CONTAINER flag is set, a container node will be
+ * created. For ZOO_*_SEQUENTIAL modes, a unique monotonically increasing
+ * sequence number is appended to the path name. The sequence number is always fixed
+ * length of 10 digits, 0 padded.
  *
  * \param zh the zookeeper handle obtained by a call to \ref zookeeper_init
  * \param path The name of the node. Expressed as a file name with slashes
@@ -1646,8 +1781,7 @@ ZOOAPI int zoo_create(zhandle_t *zh, const char *path, const char *value,
  * \param valuelen The number of bytes in data. To set the data to be NULL use
  * value as NULL and valuelen as -1.
  * \param acl The initial ACL of the node. The ACL must not be null or empty.
- * \param flags this parameter can be set to 0 for normal create or an OR
- *    of the Create Flags
+ * \param mode this parameter should be one of the Create Modes.
  * \param path_buffer Buffer which will be filled with the path of the
  *    new node (this might be different than the supplied path
  *    because of the ZOO_SEQUENCE flag).  The path string will always be
@@ -1669,7 +1803,53 @@ ZOOAPI int zoo_create(zhandle_t *zh, const char *path, const char *value,
  * ZMARSHALLINGERROR - failed to marshall a request; possibly, out of memory
  */
 ZOOAPI int zoo_create2(zhandle_t *zh, const char *path, const char *value,
-        int valuelen, const struct ACL_vector *acl, int flags,
+        int valuelen, const struct ACL_vector *acl, int mode,
+        char *path_buffer, int path_buffer_len, struct Stat *stat);
+
+/**
+ * \brief create a node synchronously and collect stat details.
+ *
+ * This method will create a node in ZooKeeper. A node can only be created if
+ * it does not already exist. The Create Mode affects the creation of nodes.
+ * If ZOO_EPHEMERAL mode is chosen, the node will automatically get removed if the
+ * client session goes away. If ZOO_CONTAINER flag is set, a container node will be
+ * created. For ZOO_*_SEQUENTIAL modes, a unique monotonically increasing
+ * sequence number is appended to the path name. The sequence number is always fixed
+ * length of 10 digits, 0 padded. When ZOO_*_WITH_TTL is selected, a ttl node will be
+ * created.
+ *
+ * \param zh the zookeeper handle obtained by a call to \ref zookeeper_init
+ * \param path The name of the node. Expressed as a file name with slashes
+ * separating ancestors of the node.
+ * \param value The data to be stored in the node.
+ * \param valuelen The number of bytes in data. To set the data to be NULL use
+ * value as NULL and valuelen as -1.
+ * \param acl The initial ACL of the node. The ACL must not be null or empty.
+ * \param mode this parameter should be one of the Create Modes.
+ * \param ttl the value of ttl in milliseconds. It must be positive for ZOO_*_WITH_TTL
+ *    Create modes, otherwise it must be -1.
+ * \param path_buffer Buffer which will be filled with the path of the
+ *    new node (this might be different than the supplied path
+ *    because of the ZOO_SEQUENCE flag).  The path string will always be
+ *    null-terminated. This parameter may be NULL if path_buffer_len = 0.
+ * \param path_buffer_len Size of path buffer; if the path of the new
+ *    node (including space for the null terminator) exceeds the buffer size,
+ *    the path string will be truncated to fit.  The actual path of the
+ *    new node in the server will not be affected by the truncation.
+ *    The path string will always be null-terminated.
+ * \param stat The Stat struct to store Stat info into.
+ * \return  one of the following codes are returned:
+ * ZOK operation completed successfully
+ * ZNONODE the parent node does not exist.
+ * ZNODEEXISTS the node already exists
+ * ZNOAUTH the client does not have permission.
+ * ZNOCHILDRENFOREPHEMERALS cannot create children of ephemeral nodes.
+ * ZBADARGUMENTS - invalid input parameters
+ * ZINVALIDSTATE - zhandle state is either ZOO_SESSION_EXPIRED_STATE or ZOO_AUTH_FAILED_STATE
+ * ZMARSHALLINGERROR - failed to marshall a request; possibly, out of memory
+ */
+ZOOAPI int zoo_create2_ttl(zhandle_t *zh, const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl, int mode, int64_t ttl,
         char *path_buffer, int path_buffer_len, struct Stat *stat);
 
 /**
@@ -1693,7 +1873,6 @@ ZOOAPI int zoo_create2(zhandle_t *zh, const char *path, const char *value,
  */
 ZOOAPI int zoo_delete(zhandle_t *zh, const char *path, int version);
 
-
 /**
  * \brief checks the existence of a node in zookeeper synchronously.
  *

zookeeper-client/zookeeper-client-c/src/cli.c

@@ -334,8 +334,8 @@ void processline(char *line) {
         line++;
     }
     if (startsWith(line, "help")) {
-      fprintf(stderr, "    create [+[e|s]] <path>\n");
-      fprintf(stderr, "    create2 [+[e|s]] <path>\n");
+      fprintf(stderr, "    create [+[e|s|c|t=ttl]] <path>\n");
+      fprintf(stderr, "    create2 [+[e|s|c|t=ttl]] <path>\n");
       fprintf(stderr, "    delete <path>\n");
       fprintf(stderr, "    set <path> <data>\n");
       fprintf(stderr, "    get <path>\n");
@@ -532,26 +532,87 @@ void processline(char *line) {
             fprintf(stderr, "Error %d for %s\n", rc, line);
         }
     } else if (startsWith(line, "create ") || startsWith(line, "create2 ")) {
-        int flags = 0;
+        int mode = 0;
+        int64_t ttl_value = -1;
         int is_create2 = startsWith(line, "create2 ");
         line += is_create2 ? 8 : 7;
+
         if (line[0] == '+') {
+            int ephemeral = 0;
+            int sequential = 0;
+            int container = 0;
+            int ttl = 0;
+
             line++;
-            if (line[0] == 'e') {
-                flags |= ZOO_EPHEMERAL;
+
+            while (*line != ' ' && *line != '\0') {
+                switch (*line) {
+                case 'e':
+                    ephemeral = 1;
+                    break;
+                case 's':
+                    sequential = 1;
+                    break;
+                case 'c':
+                    container = 1;
+                    break;
+                case 't':
+                    ttl = 1;
+
+                    line++;
+
+                    if (*line != '=') {
+                        fprintf(stderr, "Missing ttl value after +t\n");
+                        return;
+                    }
+
+                    line++;
+
+                    ttl_value = strtol(line, &line, 10);
+
+                    if (ttl_value <= 0) {
+                        fprintf(stderr, "ttl value must be a positive integer\n");
+                        return;
+                    }
+
+                    // move back line pointer to the last digit
+                    line--;
+
+                    break;
+                default:
+                    fprintf(stderr, "Unknown option: %c\n", *line);
+                    return;
+                }
+
                 line++;
             }
-            if (line[0] == 's') {
-                flags |= ZOO_SEQUENCE;
+
+            if (ephemeral != 0 && sequential == 0 && container == 0 && ttl == 0) {
+                mode = ZOO_EPHEMERAL;
+            } else if (ephemeral == 0 && sequential != 0 && container == 0 && ttl == 0) {
+                mode = ZOO_PERSISTENT_SEQUENTIAL;
+            } else if (ephemeral != 0 && sequential != 0 && container == 0 && ttl == 0) {
+                mode = ZOO_EPHEMERAL_SEQUENTIAL;
+            } else if (ephemeral == 0 && sequential == 0 && container != 0 && ttl == 0) {
+                mode = ZOO_CONTAINER;
+            } else if (ephemeral == 0 && sequential == 0 && container == 0 && ttl != 0) {
+                mode = ZOO_PERSISTENT_WITH_TTL;
+            } else if (ephemeral == 0 && sequential != 0 && container == 0 && ttl != 0) {
+                mode = ZOO_PERSISTENT_SEQUENTIAL_WITH_TTL;
+            } else {
+                fprintf(stderr, "Invalid mode.\n");
+                return;
+            }
+
+            if (*line == ' ') {
                 line++;
             }
-            line++;
         }
         if (line[0] != '/') {
             fprintf(stderr, "Path must start with /, found: %s\n", line);
             return;
         }
-        fprintf(stderr, "Creating [%s] node\n", line);
+        fprintf(stderr, "Creating [%s] node (mode: %d)\n", line, mode);
 //        {
 //            struct ACL _CREATE_ONLY_ACL_ACL[] = {{ZOO_PERM_CREATE, ZOO_ANYONE_ID_UNSAFE}};
 //            struct ACL_vector CREATE_ONLY_ACL = {1,_CREATE_ONLY_ACL_ACL};
@@ -559,10 +620,10 @@ void processline(char *line) {
 //                    my_string_completion, strdup(line));
 //        }
         if (is_create2) {
-          rc = zoo_acreate2(zh, line, "new", 3, &ZOO_OPEN_ACL_UNSAFE, flags,
+          rc = zoo_acreate2_ttl(zh, line, "new", 3, &ZOO_OPEN_ACL_UNSAFE, mode, ttl_value,
                 my_string_stat_completion_free_data, strdup(line));
         } else {
-          rc = zoo_acreate(zh, line, "new", 3, &ZOO_OPEN_ACL_UNSAFE, flags,
+          rc = zoo_acreate_ttl(zh, line, "new", 3, &ZOO_OPEN_ACL_UNSAFE, mode, ttl_value,
                 my_string_completion_free_data, strdup(line));
         }
         if (rc) {

zookeeper-client/zookeeper-client-c/src/zookeeper.c

@@ -97,9 +97,26 @@
 const int ZOOKEEPER_WRITE = 1 << 0;
 const int ZOOKEEPER_READ = 1 << 1;
 
-const int ZOO_EPHEMERAL = 1 << 0;
+const int ZOO_PERSISTENT = 0;
+const int ZOO_EPHEMERAL = 1;
+const int ZOO_PERSISTENT_SEQUENTIAL = 2;
+const int ZOO_EPHEMERAL_SEQUENTIAL = 3;
+const int ZOO_CONTAINER = 4;
+const int ZOO_PERSISTENT_WITH_TTL = 5;
+const int ZOO_PERSISTENT_SEQUENTIAL_WITH_TTL = 6;
+
+#define ZOOKEEPER_IS_SEQUENCE(mode) \
+    ((mode) == ZOO_PERSISTENT_SEQUENTIAL || \
+     (mode) == ZOO_EPHEMERAL_SEQUENTIAL || \
+     (mode) == ZOO_PERSISTENT_SEQUENTIAL_WITH_TTL)
+#define ZOOKEEPER_IS_TTL(mode) \
+    ((mode) == ZOO_PERSISTENT_WITH_TTL || \
+     (mode) == ZOO_PERSISTENT_SEQUENTIAL_WITH_TTL)
+
+// keep ZOO_SEQUENCE as a bitmask for compatibility reasons
 const int ZOO_SEQUENCE = 1 << 1;
-const int ZOO_CONTAINER = 1 << 2;
+
+#define ZOO_MAX_TTL 0xFFFFFFFFFFLL
 
 const int ZOO_EXPIRED_SESSION_STATE = EXPIRED_SESSION_STATE_DEF;
 const int ZOO_AUTH_FAILED_STATE = AUTH_FAILED_STATE_DEF;
@@ -109,8 +126,6 @@ const int ZOO_CONNECTED_STATE = CONNECTED_STATE_DEF;
 const int ZOO_READONLY_STATE = READONLY_STATE_DEF;
 const int ZOO_NOTCONNECTED_STATE = NOTCONNECTED_STATE_DEF;
 
-#define ZOOKEEPER_IS_CONTAINER(flags) (((flags) & ZOO_CONTAINER) == ZOO_CONTAINER)
-
 static __attribute__ ((unused)) const char* state2String(int state){
     switch(state){
     case 0:
@@ -257,7 +272,7 @@ static int disable_conn_permute=0; // permute enabled by default
 static struct sockaddr_storage *addr_rw_server = 0;
 
 static void *SYNCHRONOUS_MARKER = (void*)&SYNCHRONOUS_MARKER;
-static int isValidPath(const char* path, const int flags);
+static int isValidPath(const char* path, const int mode);
 
 static int aremove_watches(
     zhandle_t *zh, const char *path, ZooWatcherType wtype,
@@ -3268,7 +3283,7 @@ finish:
     return rc;
 }
 
-static int isValidPath(const char* path, const int flags) {
+static int isValidPath(const char* path, const int mode) {
     int len = 0;
     char lastc = '/';
     char c;
@@ -3283,7 +3298,7 @@ static int isValidPath(const char* path, const int flags) {
     return 0;
   if (len == 1) // done checking - it's the root
     return 1;
-  if (path[len - 1] == '/' && !(flags & ZOO_SEQUENCE))
+  if (path[len - 1] == '/' && !ZOOKEEPER_IS_SEQUENCE(mode))
     return 0;
 
   i = 1;
@@ -3295,12 +3310,12 @@ static int isValidPath(const char* path, const int flags) {
     } else if (c == '/' && lastc == '/') {
       return 0;
     } else if (c == '.' && lastc == '.') {
-      if (path[i-2] == '/' && (((i + 1 == len) && !(flags & ZOO_SEQUENCE))
+      if (path[i-2] == '/' && (((i + 1 == len) && !ZOOKEEPER_IS_SEQUENCE(mode))
                                || path[i+1] == '/')) {
         return 0;
       }
     } else if (c == '.') {
-      if ((path[i-1] == '/') && (((i + 1 == len) && !(flags & ZOO_SEQUENCE))
+      if ((path[i-1] == '/') && (((i + 1 == len) && !ZOOKEEPER_IS_SEQUENCE(mode))
                                  || path[i+1] == '/')) {
         return 0;
       }
@@ -3316,13 +3331,13 @@ static int isValidPath(const char* path, const int flags) {
  * REQUEST INIT HELPERS
  *---------------------------------------------------------------------------*/
 /* Common Request init helper functions to reduce code duplication */
-static int Request_path_init(zhandle_t *zh, int flags,
+static int Request_path_init(zhandle_t *zh, int mode,
         char **path_out, const char *path)
 {
     assert(path_out);
 
     *path_out = prepend_string(zh, path);
-    if (zh == NULL || !isValidPath(*path_out, flags)) {
+    if (zh == NULL || !isValidPath(*path_out, mode)) {
         free_duplicate_path(*path_out, path);
         return ZBADARGUMENTS;
     }
@@ -3334,11 +3349,11 @@ static int Request_path_init(zhandle_t *zh, int flags,
     return ZOK;
 }
 
-static int Request_path_watch_init(zhandle_t *zh, int flags,
+static int Request_path_watch_init(zhandle_t *zh, int mode,
         char **path_out, const char *path,
         int32_t *watch_out, uint32_t watch)
 {
-    int rc = Request_path_init(zh, flags, path_out, path);
+    int rc = Request_path_init(zh, mode, path_out, path);
     if (rc != ZOK) {
         return rc;
     }
@@ -3521,16 +3536,16 @@ int zoo_aset(zhandle_t *zh, const char *path, const char *buffer, int buflen,
 
 static int CreateRequest_init(zhandle_t *zh, struct CreateRequest *req,
         const char *path, const char *value,
-        int valuelen, const struct ACL_vector *acl_entries, int flags)
+        int valuelen, const struct ACL_vector *acl_entries, int mode)
 {
     int rc;
     assert(req);
-    rc = Request_path_init(zh, flags, &req->path, path);
+    rc = Request_path_init(zh, mode, &req->path, path);
     assert(req);
     if (rc != ZOK) {
         return rc;
     }
-    req->flags = flags;
+    req->flags = mode;
     req->data.buff = (char*)value;
     req->data.len = valuelen;
     if (acl_entries == 0) {
@@ -3543,28 +3558,99 @@ static int CreateRequest_init(zhandle_t *zh, struct CreateRequest *req,
     return ZOK;
 }
 
+static int CreateTTLRequest_init(zhandle_t *zh, struct CreateTTLRequest *req,
+        const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl_entries, int mode, int64_t ttl)
+{
+    int rc;
+    assert(req);
+    rc = Request_path_init(zh, mode, &req->path, path);
+    assert(req);
+    if (rc != ZOK) {
+        return rc;
+    }
+    req->flags = mode;
+    req->data.buff = (char*)value;
+    req->data.len = valuelen;
+    if (acl_entries == 0) {
+        req->acl.count = 0;
+        req->acl.data = 0;
+    } else {
+        req->acl = *acl_entries;
+    }
+    req->ttl = ttl;
+
+    return ZOK;
+}
+
+static int get_create_op_type(int mode, int default_op) {
+    if (mode == ZOO_CONTAINER) {
+        return ZOO_CREATE_CONTAINER_OP;
+    } else if (ZOOKEEPER_IS_TTL(mode)) {
+        return ZOO_CREATE_TTL_OP;
+    } else {
+        return default_op;
+    }
+}
+
 int zoo_acreate(zhandle_t *zh, const char *path, const char *value,
-        int valuelen, const struct ACL_vector *acl_entries, int flags,
+        int valuelen, const struct ACL_vector *acl_entries, int mode,
+        string_completion_t completion, const void *data)
+{
+    return zoo_acreate_ttl(zh, path, value, valuelen, acl_entries, mode, -1, completion, data);
+}
+
+int zoo_acreate_ttl(zhandle_t *zh, const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl_entries, int mode, int64_t ttl,
         string_completion_t completion, const void *data)
 {
     struct oarchive *oa;
-    struct RequestHeader h = {get_xid(), ZOOKEEPER_IS_CONTAINER(flags) ? ZOO_CREATE_CONTAINER_OP : ZOO_CREATE_OP};
-    struct CreateRequest req;
+    struct RequestHeader h = {get_xid(), get_create_op_type(mode, ZOO_CREATE_OP)};
+    int rc;
+    char *req_path;
 
-    int rc = CreateRequest_init(zh, &req,
-            path, value, valuelen, acl_entries, flags);
-    if (rc != ZOK) {
-        return rc;
+    if (ZOOKEEPER_IS_TTL(mode)) {
+        struct CreateTTLRequest req;
+
+        if (ttl <= 0 || ttl > ZOO_MAX_TTL) {
+            return ZBADARGUMENTS;
+        }
+
+        rc = CreateTTLRequest_init(zh, &req,
+                path, value, valuelen, acl_entries, mode, ttl);
+        if (rc != ZOK) {
+            return rc;
+        }
+        oa = create_buffer_oarchive();
+        rc = serialize_RequestHeader(oa, "header", &h);
+        rc = rc < 0 ? rc : serialize_CreateTTLRequest(oa, "req", &req);
+
+        req_path = req.path;
+    } else {
+        struct CreateRequest req;
+
+        if (ttl >= 0) {
+            return ZBADARGUMENTS;
+        }
+
+        rc = CreateRequest_init(zh, &req,
+                path, value, valuelen, acl_entries, mode);
+        if (rc != ZOK) {
+            return rc;
+        }
+        oa = create_buffer_oarchive();
+        rc = serialize_RequestHeader(oa, "header", &h);
+        rc = rc < 0 ? rc : serialize_CreateRequest(oa, "req", &req);
+
+        req_path = req.path;
     }
-    oa = create_buffer_oarchive();
-    rc = serialize_RequestHeader(oa, "header", &h);
-    rc = rc < 0 ? rc : serialize_CreateRequest(oa, "req", &req);
+
     enter_critical(zh);
     rc = rc < 0 ? rc : add_string_completion(zh, h.xid, completion, data);
     rc = rc < 0 ? rc : queue_buffer_bytes(&zh->to_send, get_buffer(oa),
             get_buffer_len(oa));
     leave_critical(zh);
-    free_duplicate_path(req.path, path);
+    free_duplicate_path(req_path, path);
     /* We queued the buffer, so don't free it */
     close_buffer_oarchive(&oa, 0);
 
@@ -3576,26 +3662,62 @@ int zoo_acreate(zhandle_t *zh, const char *path, const char *value,
 }
 
 int zoo_acreate2(zhandle_t *zh, const char *path, const char *value,
-        int valuelen, const struct ACL_vector *acl_entries, int flags,
+        int valuelen, const struct ACL_vector *acl_entries, int mode,
+        string_stat_completion_t completion, const void *data)
+{
+    return zoo_acreate2_ttl(zh, path, value, valuelen, acl_entries, mode, -1, completion, data);
+}
+
+int zoo_acreate2_ttl(zhandle_t *zh, const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl_entries, int mode, int64_t ttl,
         string_stat_completion_t completion, const void *data)
 {
     struct oarchive *oa;
-    struct RequestHeader h = { get_xid(), ZOOKEEPER_IS_CONTAINER(flags) ? ZOO_CREATE_CONTAINER_OP : ZOO_CREATE2_OP };
-    struct CreateRequest req;
+    struct RequestHeader h = { get_xid(), get_create_op_type(mode, ZOO_CREATE2_OP) };
+    int rc;
+    char *req_path;
 
-    int rc = CreateRequest_init(zh, &req, path, value, valuelen, acl_entries, flags);
-    if (rc != ZOK) {
-        return rc;
+    if (ZOOKEEPER_IS_TTL(mode)) {
+        struct CreateTTLRequest req;
+
+        if (ttl <= 0 || ttl > ZOO_MAX_TTL) {
+            return ZBADARGUMENTS;
+        }
+
+        rc = CreateTTLRequest_init(zh, &req,
+                path, value, valuelen, acl_entries, mode, ttl);
+        if (rc != ZOK) {
+            return rc;
+        }
+        oa = create_buffer_oarchive();
+        rc = serialize_RequestHeader(oa, "header", &h);
+        rc = rc < 0 ? rc : serialize_CreateTTLRequest(oa, "req", &req);
+
+        req_path = req.path;
+    } else {
+        struct CreateRequest req;
+
+        if (ttl >= 0) {
+            return ZBADARGUMENTS;
+        }
+
+        rc = CreateRequest_init(zh, &req, path, value, valuelen, acl_entries, mode);
+        if (rc != ZOK) {
+            return rc;
+        }
+        oa = create_buffer_oarchive();
+        rc = serialize_RequestHeader(oa, "header", &h);
+        rc = rc < 0 ? rc : serialize_CreateRequest(oa, "req", &req);
+
+        req_path = req.path;
     }
-    oa = create_buffer_oarchive();
-    rc = serialize_RequestHeader(oa, "header", &h);
-    rc = rc < 0 ? rc : serialize_CreateRequest(oa, "req", &req);
+
     enter_critical(zh);
     rc = rc < 0 ? rc : add_string_stat_completion(zh, h.xid, completion, data);
     rc = rc < 0 ? rc : queue_buffer_bytes(&zh->to_send, get_buffer(oa),
             get_buffer_len(oa));
     leave_critical(zh);
-    free_duplicate_path(req.path, path);
+    free_duplicate_path(req_path, path);
     /* We queued the buffer, so don't free it */
     close_buffer_oarchive(&oa, 0);
 
@@ -4124,31 +4246,32 @@ done:
     return rc;
 }
 void zoo_create_op_init(zoo_op_t *op, const char *path, const char *value,
-        int valuelen,  const struct ACL_vector *acl, int flags,
+        int valuelen,  const struct ACL_vector *acl, int mode,
         char *path_buffer, int path_buffer_len)
 {
     assert(op);
-    op->type = ZOOKEEPER_IS_CONTAINER(flags) ? ZOO_CREATE_CONTAINER_OP : ZOO_CREATE_OP;
+    op->type = get_create_op_type(mode, ZOO_CREATE_OP);
     op->create_op.path = path;
     op->create_op.data = value;
     op->create_op.datalen = valuelen;
     op->create_op.acl = acl;
-    op->create_op.flags = flags;
+    op->create_op.flags = mode;
+    op->create_op.ttl = 0;
     op->create_op.buf = path_buffer;
     op->create_op.buflen = path_buffer_len;
 }
 
 void zoo_create2_op_init(zoo_op_t *op, const char *path, const char *value,
-        int valuelen,  const struct ACL_vector *acl, int flags,
+        int valuelen,  const struct ACL_vector *acl, int mode,
         char *path_buffer, int path_buffer_len)
 {
     assert(op);
-    op->type = ZOOKEEPER_IS_CONTAINER(flags) ? ZOO_CREATE_CONTAINER_OP : ZOO_CREATE2_OP;
+    op->type = get_create_op_type(mode, ZOO_CREATE2_OP);
     op->create_op.path = path;
     op->create_op.data = value;
     op->create_op.datalen = valuelen;
     op->create_op.acl = acl;
-    op->create_op.flags = flags;
+    op->create_op.flags = mode;
     op->create_op.buf = path_buffer;
     op->create_op.buflen = path_buffer_len;
 }
@@ -4551,7 +4674,15 @@ static void process_sync_completion(zhandle_t *zh,
  * SYNC API
  *---------------------------------------------------------------------------*/
 int zoo_create(zhandle_t *zh, const char *path, const char *value,
-        int valuelen, const struct ACL_vector *acl, int flags,
+        int valuelen, const struct ACL_vector *acl, int mode,
+        char *path_buffer, int path_buffer_len)
+{
+    return zoo_create_ttl(zh, path, value, valuelen, acl, mode, -1,
+        path_buffer, path_buffer_len);
+}
+
+int zoo_create_ttl(zhandle_t *zh, const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl, int mode, int64_t ttl,
         char *path_buffer, int path_buffer_len)
 {
     struct sync_completion *sc = alloc_sync_completion();
@@ -4561,7 +4692,7 @@ int zoo_create(zhandle_t *zh, const char *path, const char *value,
     }
     sc->u.str.str = path_buffer;
     sc->u.str.str_len = path_buffer_len;
-    rc=zoo_acreate(zh, path, value, valuelen, acl, flags, SYNCHRONOUS_MARKER, sc);
+    rc=zoo_acreate_ttl(zh, path, value, valuelen, acl, mode, ttl, SYNCHRONOUS_MARKER, sc);
     if(rc==ZOK){
         wait_sync_completion(sc);
         rc = sc->rc;
@@ -4571,7 +4702,15 @@ int zoo_create(zhandle_t *zh, const char *path, const char *value,
 }
 
 int zoo_create2(zhandle_t *zh, const char *path, const char *value,
-        int valuelen, const struct ACL_vector *acl, int flags,
+        int valuelen, const struct ACL_vector *acl, int mode,
+        char *path_buffer, int path_buffer_len, struct Stat *stat)
+{
+    return zoo_create2_ttl(zh, path, value, valuelen, acl, mode, -1,
+        path_buffer, path_buffer_len, stat);
+}
+
+int zoo_create2_ttl(zhandle_t *zh, const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl, int mode, int64_t ttl,
         char *path_buffer, int path_buffer_len, struct Stat *stat)
 {
     struct sync_completion *sc = alloc_sync_completion();
@@ -4582,7 +4721,7 @@ int zoo_create2(zhandle_t *zh, const char *path, const char *value,
 
     sc->u.str.str = path_buffer;
     sc->u.str.str_len = path_buffer_len;
-    rc=zoo_acreate2(zh, path, value, valuelen, acl, flags, SYNCHRONOUS_MARKER, sc);
+    rc=zoo_acreate2_ttl(zh, path, value, valuelen, acl, mode, ttl, SYNCHRONOUS_MARKER, sc);
     if(rc==ZOK){
         wait_sync_completion(sc);
         rc = sc->rc;

zookeeper-client/zookeeper-client-c/tests/TestClient.cc

@@ -212,6 +212,7 @@ class Zookeeper_simpleSystem : public CPPUNIT_NS::TestFixture
 #endif
     CPPUNIT_TEST(testCreate);
     CPPUNIT_TEST(testCreateContainer);
+    CPPUNIT_TEST(testCreateTtl);
     CPPUNIT_TEST(testPath);
     CPPUNIT_TEST(testPathValidation);
     CPPUNIT_TEST(testPing);
@@ -723,6 +724,24 @@ public:
         CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
     }
 
+    void testCreateTtl() {
+        watchctx_t ctx;
+        int rc = 0;
+        zhandle_t *zk = createClient(&ctx);
+        CPPUNIT_ASSERT(zk);
+        char pathbuf[80];
+        struct Stat stat = {0};
+
+        rc = zoo_create2_ttl(zk, "/testTtl", "", 0, &ZOO_OPEN_ACL_UNSAFE,
+                             ZOO_PERSISTENT_WITH_TTL, 1, pathbuf, sizeof(pathbuf), &stat);
+        CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
+
+        sleep(1);
+
+        rc = zoo_exists(zk, "/testTtl", 1, &stat);
+        CPPUNIT_ASSERT_EQUAL((int) ZNONODE, rc);
+    }
+
     void testGetChildren2() {
         int rc;
         watchctx_t ctx;

zookeeper-client/zookeeper-client-c/tests/zkServer.sh

@@ -108,17 +108,19 @@ then
     CLASSPATH=`cygpath -wp "$CLASSPATH"`
 fi
 
+PROPERTIES="-Dzookeeper.extendedTypesEnabled=true -Dznode.container.checkIntervalMs=100"
+
 case $1 in
 start|startClean)
     if [ "x${base_dir}" == "x" ]
     then
         mkdir -p /tmp/zkdata
-        java -cp "$CLASSPATH" org.apache.zookeeper.server.ZooKeeperServerMain $ZOOPORT /tmp/zkdata 3000 $ZKMAXCNXNS &> /tmp/zk.log &
+        java -cp "$CLASSPATH" $PROPERTIES org.apache.zookeeper.server.ZooKeeperServerMain $ZOOPORT /tmp/zkdata 3000 $ZKMAXCNXNS &> /tmp/zk.log &
         pid=$!
         echo -n $! > /tmp/zk.pid
     else
         mkdir -p "${base_dir}/build/tmp/zkdata"
-        java -cp "$CLASSPATH" org.apache.zookeeper.server.ZooKeeperServerMain $ZOOPORT "${base_dir}/build/tmp/zkdata" 3000 $ZKMAXCNXNS &> "${base_dir}/build/tmp/zk.log" &
+        java -cp "$CLASSPATH" $PROPERTIES org.apache.zookeeper.server.ZooKeeperServerMain $ZOOPORT "${base_dir}/build/tmp/zkdata" 3000 $ZKMAXCNXNS &> "${base_dir}/build/tmp/zk.log" &
         pid=$!
         echo -n $pid > "${base_dir}/build/tmp/zk.pid"
     fi
@@ -130,7 +132,7 @@ start|startClean)
     do
         if ps -p $pid > /dev/null
         then
-            java -cp "$CLASSPATH" org.apache.zookeeper.ZooKeeperMain -server localhost:$ZOOPORT ls / > /dev/null 2>&1
+            java -cp "$CLASSPATH" $PROPERTIES org.apache.zookeeper.ZooKeeperMain -server localhost:$ZOOPORT ls / > /dev/null 2>&1
             if [ $? -ne 0  ]
             then
                 # server not up yet - wait
@@ -170,7 +172,8 @@ startReadOnly)
         sed "s#TMPDIR#${tmpdir}#g" ${base_dir}/zookeeper-client/zookeeper-client-c/tests/quorum.cfg > "${tmpdir}/quorum.cfg"
 
         # force read-only mode
-        java -cp "$CLASSPATH" -Dreadonlymode.enabled=true org.apache.zookeeper.server.quorum.QuorumPeerMain ${tmpdir}/quorum.cfg &> "${tmpdir}/zk.log" &
+	PROPERTIES="$PROPERTIES -Dreadonlymode.enabled=true"
+        java -cp "$CLASSPATH" $PROPERTIES org.apache.zookeeper.server.quorum.QuorumPeerMain ${tmpdir}/quorum.cfg &> "${tmpdir}/zk.log" &
         pid=$!
         echo -n $pid > "${base_dir}/build/tmp/zk.pid"
         sleep 3 # wait until read-only server is up