rdhabalia closed pull request #1902: Support function registration with
package-url
URL: https://github.com/apache/incubator-pulsar/pull/1902
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pom.xml b/pom.xml
index fa3d4b76c3..c5944c5f07 100644
--- a/pom.xml
+++ b/pom.xml
@@ -387,6 +387,12 @@ flexible messaging model and an intuitive client
API.</description>
<artifactId>commons-configuration</artifactId>
<version>1.6</version>
</dependency>
+
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <version>2.5</version>
+ </dependency>
<dependency>
<groupId>net.jpountz.lz4</groupId>
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
index c22b6117b2..a00a2dd408 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
@@ -74,12 +74,13 @@ public Response registerFunction(final @PathParam("tenant")
String tenant,
final @PathParam("functionName") String
functionName,
final @FormDataParam("data") InputStream
uploadedInputStream,
final @FormDataParam("data")
FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String
functionPkgUrl,
final @FormDataParam("functionDetails")
String functionDetailsJson) {
return functions.registerFunction(
- tenant, namespace, functionName, uploadedInputStream, fileDetail,
functionDetailsJson);
+ tenant, namespace, functionName, uploadedInputStream, fileDetail,
functionPkgUrl, functionDetailsJson);
}
-
+
@PUT
@ApiOperation(value = "Updates a Pulsar Function currently running in
cluster mode")
@ApiResponses(value = {
diff --git
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
index 7c2d907f55..a86036c194 100644
---
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
+++
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
@@ -76,7 +76,7 @@
FunctionDetails getFunction(String tenant, String namespace, String
function) throws PulsarAdminException;
/**
- * Create a new function.
+ * Create a new function.
*
* @param functionDetails
* the function configuration object
@@ -85,6 +85,22 @@
* Unexpected error
*/
void createFunction(FunctionDetails functionDetails, String fileName)
throws PulsarAdminException;
+
+ /**
+ * <pre>
+ * Create a new function by providing url from which fun-pkg can be
downloaded. supported url: http/file
+ * eg:
+ * File: file:/dir/fileName.jar
+ * Http: http://www.repo.com/fileName.jar
+ * </pre>
+ *
+ * @param functionDetails
+ * the function configuration object
+ * @param pkgUrl
+ * url from which pkg can be downloaded
+ * @throws PulsarAdminException
+ */
+ void createFunctionWithUrl(FunctionDetails functionDetails, String pkgUrl)
throws PulsarAdminException;
/**
* Update the configuration for a function.
diff --git
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
index 1e2ae22cd4..1a1ab60a8a 100644
---
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
+++
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
@@ -121,6 +121,23 @@ public void createFunction(FunctionDetails
functionDetails, String fileName) thr
}
}
+ @Override
+ public void createFunctionWithUrl(FunctionDetails functionDetails, String
pkgUrl) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ mp.bodyPart(new FormDataBodyPart("url", pkgUrl,
MediaType.TEXT_PLAIN_TYPE));
+
+ mp.bodyPart(new FormDataBodyPart("functionDetails",
+ printJson(functionDetails),
+ MediaType.APPLICATION_JSON_TYPE));
+
request(functions.path(functionDetails.getTenant()).path(functionDetails.getNamespace()).path(functionDetails.getName()))
+ .post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA),
ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
@Override
public void deleteFunction(String cluster, String namespace, String
function) throws PulsarAdminException {
try {
diff --git a/pulsar-functions/worker/pom.xml b/pulsar-functions/worker/pom.xml
index 72088610ea..4860ae7826 100644
--- a/pulsar-functions/worker/pom.xml
+++ b/pulsar-functions/worker/pom.xml
@@ -97,6 +97,12 @@
<groupId>org.apache.distributedlog</groupId>
<artifactId>distributedlog-core</artifactId>
</dependency>
+
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ </dependency>
+
</dependencies>
</project>
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
index 4edee60040..86de82fb45 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
@@ -35,8 +35,10 @@
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.runtime.RuntimeSpawner;
import org.apache.pulsar.functions.utils.FunctionDetailsUtils;
+import static
org.apache.pulsar.functions.worker.rest.api.FunctionsImpl.isFunctionPackageUrlSupported;
import java.io.File;
+import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.util.UUID;
import java.util.concurrent.LinkedBlockingQueue;
@@ -103,18 +105,44 @@ public void join() throws InterruptedException {
}
private void startFunction(FunctionRuntimeInfo functionRuntimeInfo) throws
Exception {
- Function.Instance instance = functionRuntimeInfo.getFunctionInstance();
- FunctionMetaData functionMetaData = instance.getFunctionMetaData();
+ FunctionMetaData functionMetaData =
functionRuntimeInfo.getFunctionInstance().getFunctionMetaData();
+ int instanceId =
functionRuntimeInfo.getFunctionInstance().getInstanceId();
log.info("Starting function {} - {} ...",
- functionMetaData.getFunctionDetails().getName(),
instance.getInstanceId());
+ functionMetaData.getFunctionDetails().getName(), instanceId);
+ File pkgFile = null;
+
+ String pkgLocation =
functionMetaData.getPackageLocation().getPackagePath();
+ boolean isPkgUrlProvided = isFunctionPackageUrlSupported(pkgLocation);
+
+ if(isPkgUrlProvided && pkgLocation.startsWith(Utils.FILE)) {
+ pkgFile = new File(pkgLocation);
+ } else {
+ downloadFile(pkgFile, isPkgUrlProvided, functionMetaData,
instanceId);
+ }
+
+ InstanceConfig instanceConfig = new InstanceConfig();
+
instanceConfig.setFunctionDetails(functionMetaData.getFunctionDetails());
+ // TODO: set correct function id and version when features implemented
+ instanceConfig.setFunctionId(UUID.randomUUID().toString());
+ instanceConfig.setFunctionVersion(UUID.randomUUID().toString());
+ instanceConfig.setInstanceId(String.valueOf(instanceId));
+ instanceConfig.setMaxBufferedTuples(1024);
+
instanceConfig.setPort(org.apache.pulsar.functions.utils.Utils.findAvailablePort());
+ RuntimeSpawner runtimeSpawner = new RuntimeSpawner(instanceConfig,
pkgFile.getAbsolutePath(),
+ runtimeFactory, workerConfig.getInstanceLivenessCheckFreqMs());
+
+ functionRuntimeInfo.setRuntimeSpawner(runtimeSpawner);
+ runtimeSpawner.start();
+ }
+
+ private void downloadFile(File pkgFile, boolean isPkgUrlProvided,
FunctionMetaData functionMetaData, int instanceId) throws
FileNotFoundException, IOException {
+
File pkgDir = new File(
workerConfig.getDownloadDirectory(),
- getDownloadPackagePath(functionMetaData,
instance.getInstanceId()));
+ getDownloadPackagePath(functionMetaData, instanceId));
pkgDir.mkdirs();
- int instanceId =
functionRuntimeInfo.getFunctionInstance().getInstanceId();
-
- File pkgFile = new File(
+ pkgFile = new File(
pkgDir,
new
File(FunctionDetailsUtils.getDownloadFileName(functionMetaData.getFunctionDetails())).getName());
@@ -133,14 +161,21 @@ private void startFunction(FunctionRuntimeInfo
functionRuntimeInfo) throws Excep
break;
}
}
- try {
- log.info("Function package file {} will be downloaded from {}",
- tempPkgFile, functionMetaData.getPackageLocation());
+ String pkgLocationPath =
functionMetaData.getPackageLocation().getPackagePath();
+ boolean downloadFromHttp = isPkgUrlProvided &&
pkgLocationPath.startsWith(Utils.HTTP);
+ log.info("Function package file {} will be downloaded from {}",
tempPkgFile,
+ downloadFromHttp ? pkgLocationPath :
functionMetaData.getPackageLocation());
+
+ if(downloadFromHttp) {
+ Utils.downloadFromHttpUrl(pkgLocationPath, new
FileOutputStream(tempPkgFile));
+ } else {
Utils.downloadFromBookkeeper(
dlogNamespace,
new FileOutputStream(tempPkgFile),
- functionMetaData.getPackageLocation().getPackagePath());
-
+ pkgLocationPath);
+ }
+
+ try {
// create a hardlink, if there are two concurrent createLink
operations, one will fail.
// this ensures one instance will successfully download the
package.
try {
@@ -157,20 +192,6 @@ private void startFunction(FunctionRuntimeInfo
functionRuntimeInfo) throws Excep
} finally {
tempPkgFile.delete();
}
-
- InstanceConfig instanceConfig = new InstanceConfig();
-
instanceConfig.setFunctionDetails(functionMetaData.getFunctionDetails());
- // TODO: set correct function id and version when features implemented
- instanceConfig.setFunctionId(UUID.randomUUID().toString());
- instanceConfig.setFunctionVersion(UUID.randomUUID().toString());
- instanceConfig.setInstanceId(String.valueOf(instanceId));
- instanceConfig.setMaxBufferedTuples(1024);
-
instanceConfig.setPort(org.apache.pulsar.functions.utils.Utils.findAvailablePort());
- RuntimeSpawner runtimeSpawner = new RuntimeSpawner(instanceConfig,
pkgFile.getAbsolutePath(),
- runtimeFactory, workerConfig.getInstanceLivenessCheckFreqMs());
-
- functionRuntimeInfo.setRuntimeSpawner(runtimeSpawner);
- runtimeSpawner.start();
}
private void stopFunction(FunctionRuntimeInfo functionRuntimeInfo) {
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
index b64a7fad2e..586938c3da 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
@@ -20,16 +20,20 @@
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
import java.util.UUID;
-
import lombok.extern.slf4j.Slf4j;
-
import org.apache.distributedlog.AppendOnlyStreamWriter;
import org.apache.distributedlog.DistributedLogConfiguration;
import org.apache.distributedlog.api.DistributedLogManager;
@@ -39,14 +43,17 @@
import org.apache.distributedlog.metadata.DLMetadata;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.worker.dlog.DLInputStream;
import org.apache.pulsar.functions.worker.dlog.DLOutputStream;
import org.apache.zookeeper.KeeperException.Code;
+import org.apache.pulsar.functions.proto.Function;
@Slf4j
public final class Utils {
+ public static String HTTP = "http";
+ public static String FILE = "file";
+
private Utils(){}
public static Object getObject(byte[] byteArr) throws IOException,
ClassNotFoundException {
@@ -124,6 +131,34 @@ public static void uploadToBookeeper(Namespace
dlogNamespace,
}
}
+ public static void validateFileUrl(String destPkgUrl, String
downloadPkgDir) throws IOException, URISyntaxException {
+ if (destPkgUrl.startsWith(FILE)) {
+ URL url = new URL(destPkgUrl);
+ File file = new File(url.toURI());
+ if (!file.exists()) {
+ throw new IOException(destPkgUrl + " does not exists locally");
+ }
+ } else if (destPkgUrl.startsWith("http")) {
+ URL website = new URL(destPkgUrl);
+ File tempFile = new File(downloadPkgDir, website.getHost() +
UUID.randomUUID().toString());
+ ReadableByteChannel rbc =
Channels.newChannel(website.openStream());
+ try (FileOutputStream fos = new FileOutputStream(tempFile)) {
+ fos.getChannel().transferFrom(rbc, 0, 10);
+ }
+ if (tempFile.exists()) {
+ tempFile.delete();
+ }
+ } else {
+ throw new IllegalArgumentException("Unsupported url protocol "+
destPkgUrl +", supported url protocols: [file/http/https]");
+ }
+ }
+
+ public static void downloadFromHttpUrl(String destPkgUrl, FileOutputStream
outputStream) throws IOException {
+ URL website = new URL(destPkgUrl);
+ ReadableByteChannel rbc = Channels.newChannel(website.openStream());
+ outputStream.getChannel().transferFrom(rbc, 0, Long.MAX_VALUE);
+ }
+
public static void downloadFromBookkeeper(Namespace namespace,
OutputStream outputStream,
String packagePath) throws
IOException {
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
index 6d5cd1ae20..cb2da74cd4 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
@@ -21,9 +21,15 @@
import static com.google.common.base.Preconditions.checkNotNull;
import com.google.gson.Gson;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.net.URISyntaxException;
+import java.net.URL;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -42,6 +48,8 @@
import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.StreamingOutput;
import lombok.extern.slf4j.Slf4j;
+
+import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
@@ -55,6 +63,7 @@
import org.apache.pulsar.functions.proto.Function.PackageLocationMetaData;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
+import org.apache.pulsar.functions.worker.FunctionActioner;
import org.apache.pulsar.functions.worker.FunctionMetaDataManager;
import org.apache.pulsar.functions.worker.FunctionRuntimeManager;
import org.apache.pulsar.functions.worker.MembershipManager;
@@ -101,6 +110,7 @@ public Response registerFunction(final @PathParam("tenant")
String tenant,
final @PathParam("functionName") String
functionName,
final @FormDataParam("data") InputStream
uploadedInputStream,
final @FormDataParam("data")
FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String
functionPkgUrl,
final @FormDataParam("functionDetails")
String functionDetailsJson) {
if (!isWorkerServiceAvailable()) {
@@ -108,11 +118,17 @@ public Response registerFunction(final
@PathParam("tenant") String tenant,
}
FunctionDetails functionDetails;
+ boolean isPkgUrlProvided = StringUtils.isNotBlank(functionPkgUrl);
// validate parameters
try {
- functionDetails = validateUpdateRequestParams(tenant, namespace,
functionName,
- uploadedInputStream, fileDetail, functionDetailsJson);
- } catch (IllegalArgumentException e) {
+ if(isPkgUrlProvided) {
+ functionDetails = validateUpdateRequestParams(tenant,
namespace, functionName,
+ functionPkgUrl, functionDetailsJson);
+ }else {
+ functionDetails = validateUpdateRequestParams(tenant,
namespace, functionName,
+ uploadedInputStream, fileDetail, functionDetailsJson);
+ }
+ } catch (Exception e) {
log.error("Invalid register function request @ /{}/{}/{}",
tenant, namespace, functionName, e);
return Response.status(Status.BAD_REQUEST)
@@ -136,10 +152,10 @@ public Response registerFunction(final
@PathParam("tenant") String tenant,
.setVersion(0);
PackageLocationMetaData.Builder packageLocationMetaDataBuilder =
PackageLocationMetaData.newBuilder()
- .setPackagePath(createPackagePath(tenant, namespace,
functionName, fileDetail.getFileName()));
+ .setPackagePath(isPkgUrlProvided ? functionPkgUrl :
createPackagePath(tenant, namespace, functionName, fileDetail.getFileName()));
functionMetaDataBuilder.setPackageLocation(packageLocationMetaDataBuilder);
- return updateRequest(functionMetaDataBuilder.build(),
uploadedInputStream);
+ return isPkgUrlProvided ?
updateRequest(functionMetaDataBuilder.build()) :
updateRequest(functionMetaDataBuilder.build(), uploadedInputStream);
}
@PUT
@@ -408,23 +424,22 @@ public Response listFunctions(final @PathParam("tenant")
String tenant,
return Response.status(Status.OK).entity(new
Gson().toJson(functionStateList.toArray())).build();
}
- private Response updateRequest(FunctionMetaData functionMetaData,
- InputStream uploadedInputStream) {
+ private Response updateRequest(FunctionMetaData functionMetaData,
InputStream uploadedInputStream) {
// Upload to bookkeeper
try {
log.info("Uploading function package to {}",
functionMetaData.getPackageLocation());
- Utils.uploadToBookeeper(
- worker().getDlogNamespace(),
- uploadedInputStream,
- functionMetaData.getPackageLocation().getPackagePath());
+ Utils.uploadToBookeeper(worker().getDlogNamespace(),
uploadedInputStream,
+ functionMetaData.getPackageLocation().getPackagePath());
} catch (IOException e) {
log.error("Error uploading file {}",
functionMetaData.getPackageLocation(), e);
- return Response.serverError()
- .type(MediaType.APPLICATION_JSON)
- .entity(new ErrorData(e.getMessage()))
+ return
Response.serverError().type(MediaType.APPLICATION_JSON).entity(new
ErrorData(e.getMessage()))
.build();
}
+ return updateRequest(functionMetaData);
+ }
+
+ private Response updateRequest(FunctionMetaData functionMetaData) {
// Submit to FMT
FunctionMetaDataManager functionMetaDataManager =
worker().getFunctionMetaDataManager();
@@ -622,14 +637,26 @@ public Response uploadFunction(final
@FormDataParam("data") InputStream uploaded
@GET
@Path("/download")
public Response downloadFunction(final @QueryParam("path") String path) {
- return Response.status(Status.OK).entity(
- new StreamingOutput() {
- @Override
- public void write(final OutputStream output) throws
IOException {
-
Utils.downloadFromBookkeeper(worker().getDlogNamespace(),
- output, Codec.encode(path));
+ return Response.status(Status.OK).entity(new StreamingOutput() {
+ @Override
+ public void write(final OutputStream output) throws IOException {
+ if (path.startsWith(Utils.HTTP)) {
+ URL url = new URL(path);
+ IOUtils.copy(url.openStream(), output);
+ } else if (path.startsWith(Utils.FILE)) {
+ URL url = new URL(path);
+ File file;
+ try {
+ file = new File(url.toURI());
+ IOUtils.copy(new FileInputStream(file), output);
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException("invalid file url
path: " + path);
}
- }).build();
+ } else {
+ Utils.downloadFromBookkeeper(worker().getDlogNamespace(),
output, Codec.encode(path));
+ }
+ }
+ }).build();
}
private void validateListFunctionRequestParams(String tenant, String
namespace) throws IllegalArgumentException {
@@ -683,11 +710,36 @@ private void validateDeregisterRequestParams(String
tenant,
}
}
+ private FunctionDetails validateUpdateRequestParams(String tenant, String
namespace, String functionName,
+ String functionPkgUrl, String functionDetailsJson)
+ throws IllegalArgumentException, IOException, URISyntaxException {
+ if (!isFunctionPackageUrlSupported(functionPkgUrl)) {
+ throw new IllegalArgumentException("Function Package url is not
valid. supported url (http/https/file)");
+ }
+ Utils.validateFileUrl(functionPkgUrl,
workerServiceSupplier.get().getWorkerConfig().getDownloadDirectory());
+ return validateUpdateRequestParams(tenant, namespace, functionName,
functionDetailsJson);
+ }
+
+ public static boolean isFunctionPackageUrlSupported(String functionPkgUrl)
{
+ return StringUtils.isBlank(functionPkgUrl)
+ || !(functionPkgUrl.startsWith(Utils.HTTP) ||
functionPkgUrl.startsWith(Utils.FILE));
+ }
+
+ private FunctionDetails validateUpdateRequestParams(String tenant,
+ String namespace,
+ String functionName,
+ InputStream uploadedInputStream,
+ FormDataContentDisposition fileDetail,
+ String functionDetailsJson) throws IllegalArgumentException {
+ if (uploadedInputStream == null || fileDetail == null) {
+ throw new IllegalArgumentException("Function Package is not
provided");
+ }
+ return validateUpdateRequestParams(tenant, namespace, functionName,
functionDetailsJson);
+ }
+
private FunctionDetails validateUpdateRequestParams(String tenant,
String namespace,
String functionName,
- InputStream uploadedInputStream,
- FormDataContentDisposition
fileDetail,
String functionDetailsJson)
throws IllegalArgumentException {
if (tenant == null) {
throw new IllegalArgumentException("Tenant is not provided");
@@ -698,9 +750,7 @@ private FunctionDetails validateUpdateRequestParams(String
tenant,
if (functionName == null) {
throw new IllegalArgumentException("Function Name is not
provided");
}
- if (uploadedInputStream == null || fileDetail == null) {
- throw new IllegalArgumentException("Function Package is not
provided");
- }
+
if (functionDetailsJson == null) {
throw new IllegalArgumentException("FunctionDetails is not
provided");
}
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
index 00c2a2836b..778eaaba6f 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
@@ -48,10 +48,11 @@ public Response registerFunction(final @PathParam("tenant")
String tenant,
final @PathParam("functionName") String
functionName,
final @FormDataParam("data") InputStream
uploadedInputStream,
final @FormDataParam("data")
FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String
functionPkgUrl,
final @FormDataParam("functionDetails")
String functionDetailsJson) {
return functions.registerFunction(
- tenant, namespace, functionName, uploadedInputStream, fileDetail,
functionDetailsJson);
+ tenant, namespace, functionName, uploadedInputStream, fileDetail,
functionPkgUrl, functionDetailsJson);
}
diff --git
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/UtilsTest.java
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/UtilsTest.java
new file mode 100644
index 0000000000..b2a92f4252
--- /dev/null
+++
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/UtilsTest.java
@@ -0,0 +1,80 @@
+/**
+ * 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.pulsar.functions.worker;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.util.UUID;
+
+import org.apache.pulsar.common.util.FutureUtil;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+/**
+ * Unit test of {@link Utils}.
+ */
+public class UtilsTest {
+
+ @Test
+ public void testValidateLocalFileUrl() throws Exception {
+ String fileLocation =
FutureUtil.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+ String testDir =
UtilsTest.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+ try {
+ // eg: fileLocation : /dir/fileName.jar (invalid)
+ Utils.validateFileUrl(fileLocation, testDir);
+ Assert.fail("should fail with invalid url: without protocol");
+ } catch (IllegalArgumentException ie) {
+ // Ok.. expected exception
+ }
+ String fileLocationWithProtocol = "file://" + fileLocation;
+ // eg: fileLocation : file:///dir/fileName.jar (valid)
+ Utils.validateFileUrl(fileLocationWithProtocol, testDir);
+ // eg: fileLocation : file:/dir/fileName.jar (valid)
+ fileLocationWithProtocol = "file:" + fileLocation;
+ Utils.validateFileUrl(fileLocationWithProtocol, testDir);
+ }
+
+ @Test
+ public void testValidateHttpFileUrl() throws Exception {
+
+ String jarHttpUrl =
"http://central.maven.org/maven2/org/apache/pulsar/pulsar-common/1.22.0-incubating/pulsar-common-1.22.0-incubating.jar";;
+ String testDir =
UtilsTest.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+ Utils.validateFileUrl(jarHttpUrl, testDir);
+
+ jarHttpUrl = "http://_invalidurl_.com";;
+ try {
+ // eg: fileLocation : /dir/fileName.jar (invalid)
+ Utils.validateFileUrl(jarHttpUrl, testDir);
+ Assert.fail("should fail with invalid url: without protocol");
+ } catch (Exception ie) {
+ // Ok.. expected exception
+ }
+ }
+
+ @Test
+ public void testDownloadFile() throws Exception {
+ String jarHttpUrl =
"http://central.maven.org/maven2/org/apache/pulsar/pulsar-common/1.22.0-incubating/pulsar-common-1.22.0-incubating.jar";;
+ String testDir =
UtilsTest.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+ File pkgFile = new File(testDir, UUID.randomUUID().toString());
+ Utils.downloadFromHttpUrl(jarHttpUrl, new FileOutputStream(pkgFile));
+ Assert.assertTrue(pkgFile.exists());
+ pkgFile.delete();
+ }
+
+}
diff --git
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2ResourceTest.java
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2ResourceTest.java
index 296273f977..16161e023d 100644
---
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2ResourceTest.java
+++
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2ResourceTest.java
@@ -29,16 +29,20 @@
import static org.powermock.api.mockito.PowerMockito.mockStatic;
import static org.testng.Assert.assertEquals;
-import com.google.gson.Gson;
-import com.google.common.collect.Lists;
+import java.io.File;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.io.OutputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
+import javax.ws.rs.core.StreamingOutput;
import org.apache.distributedlog.api.namespace.Namespace;
import org.apache.logging.log4j.Level;
@@ -48,13 +52,13 @@
import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import org.apache.pulsar.functions.api.utils.DefaultSerDe;
+import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.proto.Function.FunctionMetaData;
import org.apache.pulsar.functions.proto.Function.PackageLocationMetaData;
import org.apache.pulsar.functions.proto.Function.ProcessingGuarantees;
-import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.proto.Function.SinkSpec;
+import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.proto.Function.SubscriptionType;
-import org.apache.pulsar.functions.proto.Function.FunctionDetails;
-import org.apache.pulsar.functions.proto.Function.FunctionMetaData;
import org.apache.pulsar.functions.worker.FunctionMetaDataManager;
import org.apache.pulsar.functions.worker.Utils;
import org.apache.pulsar.functions.worker.WorkerConfig;
@@ -70,6 +74,9 @@
import org.testng.annotations.ObjectFactory;
import org.testng.annotations.Test;
+import com.google.common.collect.Lists;
+import com.google.gson.Gson;
+
/**
* Unit test of {@link FunctionApiV2Resource}.
*/
@@ -287,6 +294,7 @@ private void testRegisterFunctionMissingArguments(
function,
inputStream,
details,
+ null,
org.apache.pulsar.functions.utils.Utils.printJson(functionDetails));
assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
@@ -314,6 +322,7 @@ private Response registerDefaultFunction() throws
IOException {
function,
mockedInputStream,
mockedFormData,
+ null,
org.apache.pulsar.functions.utils.Utils.printJson(functionDetails));
}
@@ -923,4 +932,36 @@ public void testListFunctionsSuccess() throws Exception {
assertEquals(Status.OK.getStatusCode(), response.getStatus());
assertEquals(new Gson().toJson(functions), response.getEntity());
}
+
+ @Test
+ public void testDownloadFunctionHttpUrl() throws Exception {
+ String jarHttpUrl =
"http://central.maven.org/maven2/org/apache/pulsar/pulsar-common/1.22.0-incubating/pulsar-common-1.22.0-incubating.jar";;
+ String testDir =
FunctionApiV2ResourceTest.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+ FunctionsImpl function = new FunctionsImpl(null);
+ Response response = function.downloadFunction(jarHttpUrl);
+ StreamingOutput streamOutput = (StreamingOutput) response.getEntity();
+ File pkgFile = new File(testDir, UUID.randomUUID().toString());
+ OutputStream output = new FileOutputStream(pkgFile);
+ streamOutput.write(output);
+ Assert.assertTrue(pkgFile.exists());
+ if (pkgFile.exists()) {
+ pkgFile.delete();
+ }
+ }
+
+ @Test
+ public void testDownloadFunctionFile() throws Exception {
+ String fileLocation =
FutureUtil.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+ String testDir =
FunctionApiV2ResourceTest.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+ FunctionsImpl function = new FunctionsImpl(null);
+ Response response = function.downloadFunction("file://"+fileLocation);
+ StreamingOutput streamOutput = (StreamingOutput) response.getEntity();
+ File pkgFile = new File(testDir, UUID.randomUUID().toString());
+ OutputStream output = new FileOutputStream(pkgFile);
+ streamOutput.write(output);
+ Assert.assertTrue(pkgFile.exists());
+ if (pkgFile.exists()) {
+ pkgFile.delete();
+ }
+ }
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
