Rest Endpoint to Create ZNode (apache#12497)

pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/ZookeeperResource.java

@@ -44,6 +44,7 @@
 import javax.ws.rs.DELETE;
 import javax.ws.rs.DefaultValue;
 import javax.ws.rs.GET;
+import javax.ws.rs.POST;
 import javax.ws.rs.PUT;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
@@ -52,6 +53,7 @@
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import org.apache.commons.lang.StringUtils;
+import org.apache.helix.AccessOption;
 import org.apache.helix.zookeeper.datamodel.ZNRecord;
 import org.apache.helix.zookeeper.introspect.CodehausJacksonIntrospector;
 import org.apache.pinot.controller.api.access.AccessType;
@@ -256,6 +258,66 @@ public SuccessResponse putData(
     }
   }
 
+  @POST
+  @Path("/zk/create")
+  @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.UPDATE_ZNODE)
+  @Authenticate(AccessType.CREATE)
+  @Produces(MediaType.APPLICATION_JSON)
+  @Consumes(MediaType.APPLICATION_JSON)
+  @ApiOperation(value = "Create a node at a given path")
+  @ApiResponses(value = {
+      @ApiResponse(code = 200, message = "Success"), @ApiResponse(code = 204, message = "No Content"),
+      @ApiResponse(code = 400, message = "Bad Request"), @ApiResponse(code = 500, message = "Internal server error")
+  })
+  public SuccessResponse createNode(
+      @ApiParam(value = "Zookeeper Path, must start with /", required = true) @QueryParam("path") String path,
+      @ApiParam(value = "TTL of the node. TTL are only honoured for persistent znodes (access option = 0x40 or 0x80),"
+          + " in which case TTL should be > 0. If access option is not 0x40 or 0x80, it will be ignored, and we can "
+          + "set it to any value, or just ignore it", defaultValue = "-1") @QueryParam("ttl") @DefaultValue("-1")
+      int ttl, @ApiParam(value = "accessOption", defaultValue = "1") @QueryParam("accessOption") @DefaultValue("1")
+  int accessOption, String payload) {
+
+    path = validateAndNormalizeZKPath(path, false);
+
+    //validate ttl range
+    if ((accessOption == AccessOption.PERSISTENT_WITH_TTL
+        || accessOption == AccessOption.PERSISTENT_SEQUENTIAL_WITH_TTL) && ttl <= 0) {
+      throw new ControllerApplicationException(LOGGER, "TTL for persistent nodes should be > 0",
+          Response.Status.BAD_REQUEST);
+    }
+
+    if (StringUtils.isEmpty(payload)) {
+      throw new ControllerApplicationException(LOGGER, "Must provide payload", Response.Status.BAD_REQUEST);
+    }
+    ZNRecord znRecord;
+    try {
+      znRecord = MAPPER.readValue(payload, ZNRecord.class);
+    } catch (Exception e) {
+      throw new ControllerApplicationException(LOGGER, "Failed to deserialize the data", Response.Status.BAD_REQUEST,
+          e);
+    }
+
+    boolean result;
+    try {
+      result = _pinotHelixResourceManager.createZKNode(path, znRecord, accessOption, ttl);
+    } catch (Exception e) {
+      throw new ControllerApplicationException(LOGGER, "Failed to create znode at path: " + path,
+          Response.Status.INTERNAL_SERVER_ERROR, e);
+    }
+    if (result) {
+      return new SuccessResponse("Successfully created node at path: " + path);
+    } else {
+      //check if node already exists
+      if (_pinotHelixResourceManager.getZKStat(path) != null) {
+        throw new ControllerApplicationException(LOGGER, "ZNode already exists at path: " + path,
+            Response.Status.BAD_REQUEST);
+      } else {
+        throw new ControllerApplicationException(LOGGER, "Failed to create znode at path: " + path,
+            Response.Status.INTERNAL_SERVER_ERROR);
+      }
+    }
+  }
+
   @GET
   @Path("/zk/ls")
   @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_ZNODE)

pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java

@@ -1712,6 +1712,10 @@ public boolean setZKData(String path, ZNRecord record, int expectedVersion, int
     return _helixDataAccessor.getBaseDataAccessor().set(path, record, expectedVersion, accessOption);
   }
 
+  public boolean createZKNode(String path, ZNRecord record, int accessOption, long ttl) {
+    return _helixDataAccessor.getBaseDataAccessor().create(path, record, accessOption, ttl);
+  }
+
   public boolean deleteZKPath(String path) {
     return _helixDataAccessor.getBaseDataAccessor().remove(path, -1);
   }

pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/ControllerRequestURLBuilder.java

@@ -525,6 +525,14 @@ public String forZkPutChildren(String path) {
     return StringUtil.join("/", _baseUrl, "zk/putChildren", "?path=" + path);
   }
 
+  public String forZKCreate() {
+    return StringUtil.join("/", _baseUrl, "zk/create");
+  }
+
+  public String forZkDelete() {
+    return StringUtil.join("/", _baseUrl, "zk/delete");
+  }
+
   public String forZkGet(String path) {
     return StringUtil.join("/", _baseUrl, "zk/get", "?path=" + path);
   }