Add method to create full key path if it does not exist

meta-client/src/main/java/org/apache/helix/metaclient/api/MetaClientInterface.java

@@ -152,6 +152,21 @@ public Stat (EntryMode mode, int version, long ctime, long mtime, long etime) {
    */
   void create(final String key, final T data, final EntryMode mode);
 
+  /**
+   * Create an entry of given EntryMode with given key and data. If any parent node in the node
+   * hierarchy does not exist, then the parent node will attempt to be created. The entry will not
+   * be created if there is an existing entry with the same full key. Ephemeral nodes cannot have
+   * children, so only the final child in the created path will be ephemeral.
+   */
+  void recursiveCreate(final String key, final T Data, final EntryMode mode);
+
+  /**
+   * Create a TTL entry with given key, data, and expiry time (ttl). If any parent node in the node
+   * hierarchy does not exist, then the parent node will attempt to be created. The entry will not be created if
+   * there is an existing entry with the same full key.
+   */
+  void recursiveCreateWithTTL(String key, T data, long ttl);
+
   /**
    * Create an entry of given EntryMode with given key, data, and expiry time (ttl).
    * The entry will automatically purge when reached expiry time and has no children.

meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/ZkMetaClient.java

@@ -41,6 +41,7 @@
 import org.apache.helix.metaclient.api.OpResult;
 import org.apache.helix.metaclient.exception.MetaClientException;
 import org.apache.helix.metaclient.exception.MetaClientNoNodeException;
+import org.apache.helix.metaclient.exception.MetaClientNodeExistsException;
 import org.apache.helix.metaclient.impl.zk.adapter.ChildListenerAdapter;
 import org.apache.helix.metaclient.impl.zk.adapter.DataListenerAdapter;
 import org.apache.helix.metaclient.impl.zk.adapter.DirectChildListenerAdapter;
@@ -63,6 +64,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.helix.metaclient.impl.zk.util.ZkMetaClientUtil.getZkParentPath;
 import static org.apache.helix.metaclient.impl.zk.util.ZkMetaClientUtil.translateZkExceptionToMetaclientException;
 
 
@@ -104,7 +106,7 @@ public void create(String key, Object data) {
   }
 
   @Override
-  public void create(String key, Object data, MetaClientInterface.EntryMode mode) {
+  public void create(String key, Object data, EntryMode mode) {
 
     try {
       _zkClient.create(key, data, ZkMetaClientUtil.convertMetaClientMode(mode));
@@ -115,6 +117,42 @@ public void create(String key, Object data, MetaClientInterface.EntryMode mode)
     }
   }
 
+  @Override
+  public void recursiveCreate(String key, T data, EntryMode mode) {
+    recursiveCreateHelper(key, data, mode, -1);
+  }
+
+  @Override
+  public void recursiveCreateWithTTL(String key, T data, long ttl) {
+    recursiveCreateHelper(key, data, EntryMode.TTL, ttl);
+  }
+
+  private void recursiveCreateHelper(String key, T data, EntryMode mode, long ttl) {
+    boolean retry;
+    // Ephemeral nodes cannot have children, so we will create PERSISTENT nodes as the parents
+    EntryMode parentMode = (EntryMode.EPHEMERAL.equals(mode) ?
+        EntryMode.PERSISTENT : mode);
+
+    do {
+      retry = false;
+      try {
+        if (EntryMode.TTL.equals(mode)) {
+          createWithTTL(key, data, ttl);
+        } else {
+          create(key, data, mode);
+        }
+      // If create fails due to parent node not existing, then call recursion on the parent path
+      // and retry create call. We only accept failures due to missing parent node, so do not retry
+      // on other errors.
+      } catch (MetaClientNoNodeException e) {
+        String parentPath = getZkParentPath(key);
+        recursiveCreateHelper(parentPath, null, parentMode, ttl);
+        retry = true;
+      }
+    } while (retry);
+
+  }
+
   @Override
   public void createWithTTL(String key, T data, long ttl) {
     try {

meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/util/ZkMetaClientUtil.java

@@ -358,4 +358,13 @@ public static MetaClientException.ReturnCode translateZooKeeperCodeToMetaClientC
         return MetaClientException.ReturnCode.DB_USER_ERROR;
     }
   }
+
+  public static String getZkParentPath(String path) {
+    if (path.equals("/")) {
+      return null;
+    }
+
+    int idx = path.lastIndexOf('/');
+    return idx == 0 ? "/" : path.substring(0, idx);
+  }
 }

meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestStressZkClient.java

@@ -28,6 +28,7 @@
 import org.apache.helix.metaclient.api.*;
 import org.apache.helix.metaclient.datamodel.DataRecord;
 import org.apache.helix.metaclient.exception.MetaClientException;
+import org.apache.helix.metaclient.exception.MetaClientNodeExistsException;
 import org.apache.helix.metaclient.impl.zk.factory.ZkMetaClientConfig;
 import org.apache.helix.metaclient.recipes.lock.DataRecordSerializer;
 import org.testng.Assert;
@@ -129,6 +130,35 @@ public void testCreateAndRenewTTL() {
     Assert.assertEquals(_zkMetaClient.countDirectChildren(zkParentKey), 0);
   }
 
+  @Test
+  public void testRecursiveCreate() {
+    final String zkParentKey = "/stressZk_testRecursiveCreate";
+    _zkMetaClient.create(zkParentKey, ENTRY_STRING_VALUE);
+
+    int count = (int) Math.pow(TEST_ITERATION_COUNT, 1/3d);
+    for (int i = 0; i < count; i++) {
+
+      for (int j = 0; j < count; j++) {
+
+        for (int k = 0; k < count; k++) {
+          String key = zkParentKey + "/" + i + "/" + j + "/" + k;
+          _zkMetaClient.recursiveCreate(key, String.valueOf(k), PERSISTENT);
+          Assert.assertEquals(String.valueOf(k), _zkMetaClient.get(key));
+        }
+      }
+      try {
+        _zkMetaClient.recursiveCreate(zkParentKey + "/" + i, "should_fail", PERSISTENT);
+        Assert.fail("Should have failed due to node existing");
+      } catch (MetaClientNodeExistsException ignoredException) {
+      }
+    }
+
+    // cleanup
+    _zkMetaClient.recursiveDelete(zkParentKey);
+    Assert.assertEquals(_zkMetaClient.countDirectChildren(zkParentKey), 0);
+
+  }
+
   @Test
   public void testGet() {
     final String zkParentKey = "/stressZk_testGet";

meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestZkMetaClient.java

@@ -26,6 +26,8 @@
 import org.apache.helix.metaclient.exception.MetaClientException;
 import org.apache.helix.metaclient.api.DirectChildChangeListener;
 
+import java.io.StringWriter;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
@@ -38,7 +40,6 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.commons.lang3.NotImplementedException;
-import org.apache.helix.metaclient.api.ConnectStateChangeListener;
 import org.apache.helix.metaclient.api.DataChangeListener;
 import org.apache.helix.metaclient.api.Op;
 import org.apache.helix.metaclient.api.OpResult;
@@ -50,6 +51,7 @@
 
 import static org.apache.helix.metaclient.api.DataChangeListener.ChangeType.ENTRY_UPDATE;
 import static org.apache.helix.metaclient.api.MetaClientInterface.EntryMode.CONTAINER;
+import static org.apache.helix.metaclient.api.MetaClientInterface.EntryMode.EPHEMERAL;
 import static org.apache.helix.metaclient.api.MetaClientInterface.EntryMode.PERSISTENT;
 
 
@@ -97,6 +99,48 @@ public void testCreateTTL() {
     }
   }
 
+  @Test
+  public void testRecursiveCreate() {
+    final String path = "/Test/ZkMetaClient/_fullPath";
+
+
+    try (ZkMetaClient<String> zkMetaClient = createZkMetaClient()) {
+      zkMetaClient.connect();
+      MetaClientInterface.EntryMode mode = EPHEMERAL;
+
+      // Should succeed even if one of the parent nodes exists
+      String extendedPath = "/A" + path;
+      zkMetaClient.create("/A", ENTRY_STRING_VALUE, PERSISTENT);
+      zkMetaClient.recursiveCreate(extendedPath, ENTRY_STRING_VALUE, mode);
+      Assert.assertNotNull(zkMetaClient.exists(extendedPath));
+
+      // Should succeed if no parent nodes exist
+      zkMetaClient.recursiveCreate(path, ENTRY_STRING_VALUE, mode);
+      Assert.assertNotNull(zkMetaClient.exists(path));
+      Assert.assertEquals(zkMetaClient.getDataAndStat("/Test").getRight().getEntryType(), PERSISTENT);
+      Assert.assertEquals(zkMetaClient.getDataAndStat(path).getRight().getEntryType(), mode);
+
+      // Should throw NodeExistsException if child node exists
+      zkMetaClient.recursiveCreate(path, ENTRY_STRING_VALUE, mode);
+      Assert.fail("Should have failed due to node already created");
+    } catch (MetaClientException e) {
+      Assert.assertEquals(e.getMessage(), "org.apache.helix.zookeeper.zkclient.exception.ZkNodeExistsException: org.apache.zookeeper.KeeperException$NodeExistsException: KeeperErrorCode = NodeExists for /Test/ZkMetaClient/_fullPath");
+      System.out.println(e.getMessage());
+    }
+
+  }
+
+  @Test
+  public void testRecursiveCreateWithTTL() {
+    final String path = "/Test/ZkMetaClient/_fullPath/withTTL";
+
+    try (ZkMetaClient<String> zkMetaClient = createZkMetaClient()) {
+      zkMetaClient.connect();
+      zkMetaClient.recursiveCreateWithTTL(path, ENTRY_STRING_VALUE, 1000);
+      Assert.assertNotNull(zkMetaClient.exists(path));
+    }
+  }
+
   @Test
   public void testRenewTTL() {
     final String key = "/TestZkMetaClient_testRenewTTL_1";