You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by sa...@apache.org on 2018/10/08 22:03:06 UTC
[pulsar] branch srkukarni/serverside_validation_endpoints created
(now 8fd2209)
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a change to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
at 8fd2209 Logic correction
This branch includes the following new commits:
new ee81c7e First cut of adding new endpoints for source/sink
new 48f5233 More better check
new eacd453 Merge branch 'master' into serverside_validation_endpoints
new e59600a Merge branch 'master' into serverside_validation_endpoints
new f857765 Plugged in source/sink
new 41b108a Merge branch 'master' into serverside_validation_endpoints
new e4f848c Merge branch 'master' into serverside_validation_endpoints
new 4334d62 Made pulsar admin to use Source/Sink endpoints
new d84aab1 Merge branch 'master' into serverside_validation_endpoints
new 44de9f6 Switched cli to use functionConfig instead of function details
new 625a31b Fix unittests
new d6c62cf Fixed unittest
new 8fd2209 Logic correction
The 13 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[pulsar] 04/13: Merge branch 'master' into
serverside_validation_endpoints
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e59600aca89331b00077655fb575e94674de9c67
Merge: eacd453 525239b
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Fri Oct 5 08:57:25 2018 -0700
Merge branch 'master' into serverside_validation_endpoints
conf/functions_worker.yml | 99 +++++++++++++++++++---
.../org/apache/pulsar/PulsarBrokerStarter.java | 13 +--
.../org/apache/pulsar/broker/PulsarService.java | 24 +++++-
.../functions/runtime/KubernetesRuntime.java | 4 +-
.../functions/runtime/KubernetesRuntimeTest.java | 10 ++-
.../functions/worker/FunctionsStatsGenerator.java | 9 +-
.../worker/FunctionStatsGeneratorTest.java | 30 +++++++
7 files changed, 158 insertions(+), 31 deletions(-)
[pulsar] 05/13: Plugged in source/sink
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f8577658f940a582ee59d4e2467f1465a2655de1
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Fri Oct 5 16:23:50 2018 -0700
Plugged in source/sink
---
.../pulsar/functions/utils/SinkConfigUtils.java | 9 +--
.../pulsar/functions/utils/SourceConfigUtils.java | 12 +--
.../pulsar/functions/utils/io/ConnectorUtils.java | 84 ++++++++++-----------
.../functions/utils/validation/ValidatorImpls.java | 85 +++++++++++-----------
.../org/apache/pulsar/functions/worker/Utils.java | 28 +++++++
.../functions/worker/rest/api/FunctionsImpl.java | 60 +++++++++++++--
.../worker/rest/api/v2/SinkApiV2Resource.java | 4 +-
.../worker/rest/api/v2/SourceApiV2Resource.java | 14 ++--
8 files changed, 179 insertions(+), 117 deletions(-)
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
index d44ea30..95803ab 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
@@ -37,7 +37,7 @@ import static org.apache.pulsar.functions.utils.Utils.convertProcessingGuarantee
public class SinkConfigUtils {
- public static FunctionDetails convert(SinkConfig sinkConfig) throws IOException {
+ public static FunctionDetails convert(SinkConfig sinkConfig, NarClassLoader classLoader) throws IOException {
String sinkClassName = null;
String typeArg = null;
@@ -53,11 +53,8 @@ public class SinkConfigUtils {
}
sinkClassName = sinkConfig.getClassName(); // server derives the arg-type by loading a class
} else {
- sinkClassName = ConnectorUtils.getIOSinkClass(sinkConfig.getArchive());
- try (NarClassLoader ncl = NarClassLoader.getFromArchive(new File(sinkConfig.getArchive()),
- Collections.emptySet())) {
- typeArg = Utils.getSinkType(sinkClassName, ncl).getName();
- }
+ sinkClassName = ConnectorUtils.getIOSinkClass(classLoader);
+ typeArg = Utils.getSinkType(sinkClassName, classLoader).getName();
}
}
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
index 73b331a..a132c8a 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
@@ -26,16 +26,14 @@ import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.utils.io.ConnectorUtils;
-import java.io.File;
import java.io.IOException;
-import java.util.Collections;
import static org.apache.pulsar.functions.utils.Utils.convertProcessingGuarantee;
import static org.apache.pulsar.functions.utils.Utils.getSourceType;
public class SourceConfigUtils {
- public static FunctionDetails convert(SourceConfig sourceConfig)
+ public static FunctionDetails convert(SourceConfig sourceConfig, NarClassLoader classLoader)
throws IllegalArgumentException, IOException {
String sourceClassName = null;
@@ -52,12 +50,8 @@ public class SourceConfigUtils {
}
sourceClassName = sourceConfig.getClassName(); // server derives the arg-type by loading a class
} else {
- sourceClassName = ConnectorUtils.getIOSourceClass(sourceConfig.getArchive());
-
- try (NarClassLoader ncl = NarClassLoader.getFromArchive(new File(sourceConfig.getArchive()),
- Collections.emptySet())) {
- typeArg = getSourceType(sourceClassName, ncl).getName();
- }
+ sourceClassName = ConnectorUtils.getIOSourceClass(classLoader);
+ typeArg = getSourceType(sourceClassName, classLoader).getName();
}
}
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java
index 915df05..c6feb5a 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java
@@ -46,59 +46,57 @@ public class ConnectorUtils {
/**
* Extract the Pulsar IO Source class from a connector archive.
*/
- public static String getIOSourceClass(String narPath) throws IOException {
- try (NarClassLoader ncl = NarClassLoader.getFromArchive(new File(narPath), Collections.emptySet())) {
- String configStr = ncl.getServiceDefinition(PULSAR_IO_SERVICE_NAME);
-
- ConnectorDefinition conf = ObjectMapperFactory.getThreadLocalYaml().readValue(configStr,
- ConnectorDefinition.class);
- if (StringUtils.isEmpty(conf.getSourceClass())) {
- throw new IOException(
- String.format("The '%s' connector does not provide a source implementation", conf.getName()));
- }
+ public static String getIOSourceClass(ClassLoader classLoader) throws IOException {
+ NarClassLoader ncl = (NarClassLoader) classLoader;
+ String configStr = ncl.getServiceDefinition(PULSAR_IO_SERVICE_NAME);
+
+ ConnectorDefinition conf = ObjectMapperFactory.getThreadLocalYaml().readValue(configStr,
+ ConnectorDefinition.class);
+ if (StringUtils.isEmpty(conf.getSourceClass())) {
+ throw new IOException(
+ String.format("The '%s' connector does not provide a source implementation", conf.getName()));
+ }
- try {
- // Try to load source class and check it implements Source interface
- Object instance = ncl.loadClass(conf.getSourceClass()).newInstance();
- if (!(instance instanceof Source)) {
- throw new IOException("Class " + conf.getSourceClass() + " does not implement interface "
- + Source.class.getName());
- }
- } catch (Throwable t) {
- Exceptions.rethrowIOException(t);
+ try {
+ // Try to load source class and check it implements Source interface
+ Object instance = ncl.loadClass(conf.getSourceClass()).newInstance();
+ if (!(instance instanceof Source)) {
+ throw new IOException("Class " + conf.getSourceClass() + " does not implement interface "
+ + Source.class.getName());
}
-
- return conf.getSourceClass();
+ } catch (Throwable t) {
+ Exceptions.rethrowIOException(t);
}
+
+ return conf.getSourceClass();
}
/**
* Extract the Pulsar IO Sink class from a connector archive.
*/
- public static String getIOSinkClass(String narPath) throws IOException {
- try (NarClassLoader ncl = NarClassLoader.getFromArchive(new File(narPath), Collections.emptySet())) {
- String configStr = ncl.getServiceDefinition(PULSAR_IO_SERVICE_NAME);
+ public static String getIOSinkClass(ClassLoader classLoader) throws IOException {
+ NarClassLoader ncl = (NarClassLoader) classLoader;
+ String configStr = ncl.getServiceDefinition(PULSAR_IO_SERVICE_NAME);
+
+ ConnectorDefinition conf = ObjectMapperFactory.getThreadLocalYaml().readValue(configStr,
+ ConnectorDefinition.class);
+ if (StringUtils.isEmpty(conf.getSinkClass())) {
+ throw new IOException(
+ String.format("The '%s' connector does not provide a sink implementation", conf.getName()));
+ }
- ConnectorDefinition conf = ObjectMapperFactory.getThreadLocalYaml().readValue(configStr,
- ConnectorDefinition.class);
- if (StringUtils.isEmpty(conf.getSinkClass())) {
+ try {
+ // Try to load source class and check it implements Sink interface
+ Object instance = ncl.loadClass(conf.getSinkClass()).newInstance();
+ if (!(instance instanceof Sink)) {
throw new IOException(
- String.format("The '%s' connector does not provide a sink implementation", conf.getName()));
- }
-
- try {
- // Try to load source class and check it implements Sink interface
- Object instance = ncl.loadClass(conf.getSinkClass()).newInstance();
- if (!(instance instanceof Sink)) {
- throw new IOException(
- "Class " + conf.getSinkClass() + " does not implement interface " + Sink.class.getName());
- }
- } catch (Throwable t) {
- Exceptions.rethrowIOException(t);
+ "Class " + conf.getSinkClass() + " does not implement interface " + Sink.class.getName());
}
-
- return conf.getSinkClass();
+ } catch (Throwable t) {
+ Exceptions.rethrowIOException(t);
}
+
+ return conf.getSinkClass();
}
public static ConnectorDefinition getConnectorDefinition(String narPath) throws IOException {
@@ -127,14 +125,10 @@ public class ConnectorUtils {
log.info("Found connector {} from {}", cntDef, archive);
if (!StringUtils.isEmpty(cntDef.getSourceClass())) {
- // Validate source class to be present and of the right type
- ConnectorUtils.getIOSourceClass(archive.toString());
connectors.sources.put(cntDef.getName(), archive);
}
if (!StringUtils.isEmpty(cntDef.getSinkClass())) {
- // Validate sinkclass to be present and of the right type
- ConnectorUtils.getIOSinkClass(archive.toString());
connectors.sinks.put(cntDef.getName(), archive);
}
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
index dba6bd9..e600afe 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
@@ -727,14 +727,15 @@ public class ValidatorImpls {
@Override
public void validateField(String name, Object o, ClassLoader classLoader) {
SourceConfig sourceConfig = (SourceConfig) o;
- if (sourceConfig.getArchive().startsWith(Utils.BUILTIN)) {
- // We don't have to check the archive, since it's provided on the worker itself
+ if (classLoader == null) {
+ // This happens at the cli for builtin. There is no need to check this since
+ // the actual check will be done at serverside
return;
}
String sourceClassName;
try {
- sourceClassName = ConnectorUtils.getIOSourceClass(sourceConfig.getArchive());
+ sourceClassName = ConnectorUtils.getIOSourceClass(classLoader);
} catch (IOException e1) {
throw new IllegalArgumentException("Failed to extract source class from archive", e1);
}
@@ -743,15 +744,14 @@ public class ValidatorImpls {
Class<?> typeArg = getSourceType(sourceClassName, classLoader);
// Only one of serdeClassName or schemaType should be set
- if (sourceConfig.getSerdeClassName() != null && !sourceConfig.getSerdeClassName().isEmpty()
- && sourceConfig.getSchemaType() != null && !sourceConfig.getSchemaType().isEmpty()) {
+ if (!StringUtils.isEmpty(sourceConfig.getSerdeClassName()) && !StringUtils.isEmpty(sourceConfig.getSchemaType())) {
throw new IllegalArgumentException("Only one of serdeClassName or schemaType should be set");
}
- if (sourceConfig.getSerdeClassName() != null && !sourceConfig.getSerdeClassName().isEmpty()) {
+ if (!StringUtils.isEmpty(sourceConfig.getSerdeClassName())) {
FunctionConfigValidator.validateSerde(sourceConfig.getSerdeClassName(),typeArg, name, classLoader, false);
}
- if (sourceConfig.getSchemaType() != null && !sourceConfig.getSchemaType().isEmpty()) {
+ if (!StringUtils.isEmpty(sourceConfig.getSchemaType())) {
FunctionConfigValidator.validateSchema(sourceConfig.getSchemaType(), typeArg, name, classLoader, false);
}
}
@@ -761,8 +761,9 @@ public class ValidatorImpls {
@Override
public void validateField(String name, Object o, ClassLoader classLoader) {
SinkConfig sinkConfig = (SinkConfig) o;
- if (sinkConfig.getArchive().startsWith(Utils.BUILTIN)) {
- // We don't have to check the archive, since it's provided on the worker itself
+ if (classLoader == null) {
+ // This happens at the cli for builtin. There is no need to check this since
+ // the actual check will be done at serverside
return;
}
@@ -779,42 +780,42 @@ public class ValidatorImpls {
}
- try (NarClassLoader clsLoader = NarClassLoader.getFromArchive(new File(sinkConfig.getArchive()),
- Collections.emptySet())) {
- String sinkClassName = ConnectorUtils.getIOSinkClass(sinkConfig.getArchive());
- Class<?> typeArg = getSinkType(sinkClassName, clsLoader);
+ String sinkClassName;
+ try {
+ sinkClassName = ConnectorUtils.getIOSinkClass(classLoader);
+ } catch (IOException e1) {
+ throw new IllegalArgumentException("Failed to extract sink class from archive", e1);
+ }
+ Class<?> typeArg = getSinkType(sinkClassName, classLoader);
- if (sinkConfig.getTopicToSerdeClassName() != null) {
- sinkConfig.getTopicToSerdeClassName().forEach((topicName, serdeClassName) -> {
- FunctionConfigValidator.validateSerde(serdeClassName, typeArg, name, clsLoader, true);
- });
- }
+ if (sinkConfig.getTopicToSerdeClassName() != null) {
+ sinkConfig.getTopicToSerdeClassName().forEach((topicName, serdeClassName) -> {
+ FunctionConfigValidator.validateSerde(serdeClassName, typeArg, name, classLoader, true);
+ });
+ }
- if (sinkConfig.getTopicToSchemaType() != null) {
- sinkConfig.getTopicToSchemaType().forEach((topicName, schemaType) -> {
- FunctionConfigValidator.validateSchema(schemaType, typeArg, name, clsLoader, true);
- });
- }
+ if (sinkConfig.getTopicToSchemaType() != null) {
+ sinkConfig.getTopicToSchemaType().forEach((topicName, schemaType) -> {
+ FunctionConfigValidator.validateSchema(schemaType, typeArg, name, classLoader, true);
+ });
+ }
- // topicsPattern does not need checks
-
- if (sinkConfig.getInputSpecs() != null) {
- sinkConfig.getInputSpecs().forEach((topicName, consumerSpec) -> {
- // Only one is set
- if (consumerSpec.getSerdeClassName() != null && !consumerSpec.getSerdeClassName().isEmpty()
- && consumerSpec.getSchemaType() != null && !consumerSpec.getSchemaType().isEmpty()) {
- throw new IllegalArgumentException("Only one of serdeClassName or schemaType should be set");
- }
- if (consumerSpec.getSerdeClassName() != null && !consumerSpec.getSerdeClassName().isEmpty()) {
- FunctionConfigValidator.validateSerde(consumerSpec.getSerdeClassName(), typeArg, name, clsLoader, true);
- }
- if (consumerSpec.getSchemaType() != null && !consumerSpec.getSchemaType().isEmpty()) {
- FunctionConfigValidator.validateSchema(consumerSpec.getSchemaType(), typeArg, name, clsLoader, true);
- }
- });
- }
- } catch (IOException e) {
- throw new IllegalArgumentException(e.getMessage());
+ // topicsPattern does not need checks
+
+ if (sinkConfig.getInputSpecs() != null) {
+ sinkConfig.getInputSpecs().forEach((topicName, consumerSpec) -> {
+ // Only one is set
+ if (consumerSpec.getSerdeClassName() != null && !consumerSpec.getSerdeClassName().isEmpty()
+ && consumerSpec.getSchemaType() != null && !consumerSpec.getSchemaType().isEmpty()) {
+ throw new IllegalArgumentException("Only one of serdeClassName or schemaType should be set");
+ }
+ if (consumerSpec.getSerdeClassName() != null && !consumerSpec.getSerdeClassName().isEmpty()) {
+ FunctionConfigValidator.validateSerde(consumerSpec.getSerdeClassName(), typeArg, name, classLoader, true);
+ }
+ if (consumerSpec.getSchemaType() != null && !consumerSpec.getSchemaType().isEmpty()) {
+ FunctionConfigValidator.validateSchema(consumerSpec.getSchemaType(), typeArg, name, classLoader, true);
+ }
+ });
}
}
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 aa80403..f935991 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
@@ -33,6 +33,7 @@ import java.net.URISyntaxException;
import java.net.URL;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
+import java.util.Collections;
import java.util.UUID;
import lombok.extern.slf4j.Slf4j;
@@ -47,6 +48,7 @@ import org.apache.distributedlog.metadata.DLMetadata;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.nar.NarClassLoader;
import org.apache.pulsar.functions.utils.Reflections;
import org.apache.pulsar.functions.worker.dlog.DLInputStream;
import org.apache.pulsar.functions.worker.dlog.DLOutputStream;
@@ -163,6 +165,32 @@ public final class Utils {
throw new IllegalArgumentException("Unsupported url protocol "+ destPkgUrl +", supported url protocols: [file/http/https]");
}
}
+
+ public static NarClassLoader extractNarClassloader(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 IllegalArgumentException(destPkgUrl + " does not exists locally");
+ }
+ try {
+ return NarClassLoader.getFromArchive(file, Collections.emptySet());
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(
+ "Corrupt User PackageFile " + file + " with error " + e.getMessage());
+ }
+ } else if (destPkgUrl.startsWith("http")) {
+ URL website = new URL(destPkgUrl);
+ File tempFile = new File(downloadPkgDir, website.getHost() + UUID.randomUUID().toString());
+ if (!tempFile.exists()) {
+ throw new IllegalArgumentException("Could not create local file " + tempFile);
+ }
+ tempFile.deleteOnExit();
+ return NarClassLoader.getFromArchive(tempFile, Collections.emptySet());
+ } 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);
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 0d30e37..08cd7de 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
@@ -39,10 +39,8 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.nio.file.Files;
-import java.util.Base64;
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.List;
+import java.nio.file.Path;
+import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -75,6 +73,7 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.common.nar.NarClassLoader;
import org.apache.pulsar.common.policies.data.ErrorData;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.util.Codec;
@@ -1007,13 +1006,13 @@ public class FunctionsImpl {
}
if (!StringUtils.isEmpty(sourceConfigJson)) {
SourceConfig sourceConfig = new Gson().fromJson(sourceConfigJson, SourceConfig.class);
- ClassLoader clsLoader = extractClassLoader(functionPkgUrl, uploadedInputStreamAsFile);
+ NarClassLoader clsLoader = extractNarClassLoader(sourceConfig.getArchive(), functionPkgUrl, uploadedInputStreamAsFile, true);
ConfigValidation.validateConfig(sourceConfig, FunctionConfig.Runtime.JAVA.name(), clsLoader);
return SourceConfigUtils.convert(sourceConfig, clsLoader);
}
if (!StringUtils.isEmpty(sinkConfigJson)) {
SinkConfig sinkConfig = new Gson().fromJson(sinkConfigJson, SinkConfig.class);
- ClassLoader clsLoader = extractClassLoader(functionPkgUrl, uploadedInputStreamAsFile);
+ NarClassLoader clsLoader = extractNarClassLoader(sinkConfig.getArchive(), functionPkgUrl, uploadedInputStreamAsFile, false);
ConfigValidation.validateConfig(sinkConfig, FunctionConfig.Runtime.JAVA.name(), clsLoader);
return SinkConfigUtils.convert(sinkConfig, clsLoader);
}
@@ -1076,6 +1075,55 @@ public class FunctionsImpl {
}
}
+ private NarClassLoader extractNarClassLoader(String archive, String pkgUrl, File uploadedInputStreamFileName,
+ boolean isSource) {
+ if (!StringUtils.isEmpty(archive)) {
+ if (isSource) {
+ Path path;
+ try {
+ path = this.worker().getConnectorsManager().getSourceArchive(archive);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("No Source archive %s found", archive));
+ }
+ try {
+ return NarClassLoader.getFromArchive(path.toFile(),
+ Collections.emptySet());
+ } catch (IOException e) {
+ throw new IllegalArgumentException(String.format("The source %s is corrupted", archive));
+ }
+ } else {
+ Path path;
+ try {
+ path = this.worker().getConnectorsManager().getSinkArchive(archive);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("No Sink archive %s found", archive));
+ }
+ try {
+ return NarClassLoader.getFromArchive(path.toFile(),
+ Collections.emptySet());
+ } catch (IOException e) {
+ throw new IllegalArgumentException(String.format("The sink %s is corrupted", archive));
+ }
+ }
+ }
+ if (!StringUtils.isEmpty(pkgUrl)) {
+ try {
+ return Utils.extractNarClassloader(pkgUrl, workerServiceSupplier.get().getWorkerConfig().getDownloadDirectory());
+ } catch (Exception e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ }
+ if (uploadedInputStreamFileName != null) {
+ try {
+ return NarClassLoader.getFromArchive(uploadedInputStreamFileName,
+ Collections.emptySet());
+ } catch (IOException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ }
+ return null;
+ }
+
private void validateFunctionClassTypes(ClassLoader classLoader, FunctionDetails.Builder functionDetailsBuilder) {
// validate only if classLoader is provided
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2Resource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2Resource.java
index 175e2eb..e36b069 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2Resource.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2Resource.java
@@ -48,11 +48,11 @@ public class SinkApiV2Resource extends FunctionApiResource {
final @PathParam("sinkName") String sinkName,
final @FormDataParam("data") InputStream uploadedInputStream,
final @FormDataParam("data") FormDataContentDisposition fileDetail,
- final @FormDataParam("url") String sourcePkgUrl,
+ final @FormDataParam("url") String functionPkgUrl,
final @FormDataParam("sinkConfig") String sinkConfigJson) {
return functions.registerFunction(tenant, namespace, sinkName, uploadedInputStream, fileDetail,
- sourcePkgUrl, null, null, null, sinkConfigJson, clientAppId());
+ functionPkgUrl, null, null, null, sinkConfigJson, clientAppId());
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2Resource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2Resource.java
index 9d25bf9..cfb6baa 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2Resource.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2Resource.java
@@ -44,15 +44,15 @@ public class SourceApiV2Resource extends FunctionApiResource {
@Path("/{tenant}/{namespace}/{sourceName}")
@Consumes(MediaType.MULTIPART_FORM_DATA)
public Response registerSource(final @PathParam("tenant") String tenant,
- final @PathParam("namespace") String namespace,
- final @PathParam("sourceName") String sourceName,
- final @FormDataParam("data") InputStream uploadedInputStream,
- final @FormDataParam("data") FormDataContentDisposition fileDetail,
- final @FormDataParam("url") String sourcePkgUrl,
- final @FormDataParam("sourceConfig") String sourceConfigJson) {
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String functionPkgUrl,
+ final @FormDataParam("sourceConfig") String sourceConfigJson) {
return functions.registerFunction(tenant, namespace, sourceName, uploadedInputStream, fileDetail,
- sourcePkgUrl, null, null, sourceConfigJson, null, clientAppId());
+ functionPkgUrl, null, null, sourceConfigJson, null, clientAppId());
}
[pulsar] 06/13: Merge branch 'master' into
serverside_validation_endpoints
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 41b108a8e09a0b4aa24621a1fd6add2bcc3fa719
Merge: f857765 d26bea8
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Fri Oct 5 16:25:01 2018 -0700
Merge branch 'master' into serverside_validation_endpoints
.../instance/src/main/python/python_instance.py | 7 +-
.../src/main/python/python_instance_main.py | 8 +-
.../pulsar/functions/runtime/JavaInstanceMain.java | 3 +
.../functions/runtime/KubernetesRuntime.java | 3 +-
.../runtime/KubernetesRuntimeFactory.java | 4 +
.../pulsar/functions/runtime/ProcessRuntime.java | 2 +-
.../pulsar/functions/runtime/RuntimeSpawner.java | 17 ++--
.../pulsar/functions/runtime/RuntimeUtils.java | 7 +-
.../functions/runtime/KubernetesRuntimeTest.java | 10 +--
.../functions/worker/FunctionRuntimeManager.java | 1 +
.../pulsar/functions/worker/WorkerConfig.java | 1 +
.../integration/containers/PulsarContainer.java | 5 +-
.../containers/StandaloneContainer.java | 65 +++++++++++++++
.../tests/integration/semantics/SemanticsTest.java | 33 +-------
.../tests/integration/standalone/SmokeTest.java | 32 ++++++++
.../suites/PulsarStandaloneTestSuite.java | 43 ++++++++++
.../topologies/PulsarClusterTestBase.java | 35 +-------
.../topologies/PulsarStandaloneTestBase.java | 83 +++++++++++++++++++
.../integration/topologies/PulsarTestBase.java | 93 ++++++++++++++++++++++
.../src/test/resources/pulsar-standalone.xml | 28 +++++++
20 files changed, 398 insertions(+), 82 deletions(-)
[pulsar] 02/13: More better check
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 48f5233f3eb0975aaf37974b4aa1310f292f41f9
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Thu Oct 4 20:09:24 2018 -0700
More better check
---
.../src/main/java/org/apache/pulsar/functions/utils/SourceConfig.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfig.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfig.java
index 38e200a..c1ff240 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfig.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfig.java
@@ -25,6 +25,7 @@ import lombok.Setter;
import lombok.ToString;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.functions.api.SerDe;
+import org.apache.pulsar.functions.utils.validation.ConfigValidationAnnotations;
import org.apache.pulsar.functions.utils.validation.ConfigValidationAnnotations.NotNull;
import org.apache.pulsar.functions.utils.validation.ConfigValidationAnnotations.isFileExists;
import org.apache.pulsar.functions.utils.validation.ConfigValidationAnnotations.isImplementationOfClass;
@@ -55,7 +56,7 @@ public class SourceConfig {
@isValidTopicName
private String topicName;
- @isImplementationOfClass(implementsClass = SerDe.class)
+ @ConfigValidationAnnotations.isValidSerde
private String serdeClassName;
private String schemaType;
[pulsar] 12/13: Fixed unittest
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d6c62cf823064d3c136ee423bf0c08f192f1828d
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Mon Oct 8 14:13:45 2018 -0700
Fixed unittest
---
.../apache/pulsar/admin/cli/CmdFunctionsTest.java | 23 +++++++++++-----------
1 file changed, 12 insertions(+), 11 deletions(-)
diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java
index de5a9e0..0c512b8 100644
--- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java
+++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java
@@ -44,6 +44,7 @@ import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.sink.PulsarSink;
+import org.apache.pulsar.functions.utils.FunctionConfig;
import org.apache.pulsar.functions.utils.Reflections;
import org.apache.pulsar.functions.utils.Utils;
import org.powermock.api.mockito.PowerMockito;
@@ -213,7 +214,7 @@ public class CmdFunctionsTest {
assertEquals(outputTopicName, creater.getOutput());
assertEquals(false, creater.isAutoAck());
- verify(functions, times(1)).createFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).createFunction(any(FunctionConfig.class), anyString());
}
@@ -343,7 +344,7 @@ public class CmdFunctionsTest {
assertEquals(fnName, creater.getFunctionName());
assertEquals(inputTopicName, creater.getInputs());
assertEquals(outputTopicName, creater.getOutput());
- verify(functions, times(1)).createFunctionWithUrl(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).createFunctionWithUrl(any(FunctionConfig.class), anyString());
}
@Test
@@ -420,7 +421,7 @@ public class CmdFunctionsTest {
assertEquals(topicPatterns, creater.getTopicsPattern());
assertEquals(outputTopicName, creater.getOutput());
- verify(functions, times(1)).createFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).createFunction(any(FunctionConfig.class), anyString());
}
@@ -441,7 +442,7 @@ public class CmdFunctionsTest {
CreateFunction creater = cmd.getCreater();
assertEquals("public", creater.getFunctionConfig().getTenant());
- verify(functions, times(1)).createFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).createFunction(any(FunctionConfig.class), anyString());
}
@Test
@@ -461,7 +462,7 @@ public class CmdFunctionsTest {
CreateFunction creater = cmd.getCreater();
assertEquals("public", creater.getFunctionConfig().getTenant());
assertEquals("default", creater.getFunctionConfig().getNamespace());
- verify(functions, times(1)).createFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).createFunction(any(FunctionConfig.class), anyString());
}
@Test
@@ -486,7 +487,7 @@ public class CmdFunctionsTest {
assertEquals(tenant, creater.getFunctionConfig().getTenant());
assertEquals(namespace, creater.getFunctionConfig().getNamespace());
assertEquals(functionName, creater.getFunctionConfig().getName());
- verify(functions, times(1)).createFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).createFunction(any(FunctionConfig.class), anyString());
}
@Test
@@ -505,7 +506,7 @@ public class CmdFunctionsTest {
CreateFunction creater = cmd.getCreater();
assertEquals("CmdFunctionsTest$DummyFunction", creater.getFunctionConfig().getName());
- verify(functions, times(1)).createFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).createFunction(any(FunctionConfig.class), anyString());
}
@Test
@@ -522,7 +523,7 @@ public class CmdFunctionsTest {
CreateFunction creater = cmd.getCreater();
assertNull(creater.getFunctionConfig().getOutput());
- verify(functions, times(1)).createFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).createFunction(any(FunctionConfig.class), anyString());
}
@@ -613,7 +614,7 @@ public class CmdFunctionsTest {
assertEquals(inputTopicName, updater.getInputs());
assertEquals(outputTopicName, updater.getOutput());
- verify(functions, times(1)).updateFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).updateFunction(any(FunctionConfig.class), anyString());
}
@Test
@@ -711,9 +712,9 @@ public class CmdFunctionsTest {
}
if (type.equals("create")) {
- verify(functions, times(1)).createFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).createFunction(any(FunctionConfig.class), anyString());
} else if (type.equals("update")) {
- verify(functions, times(1)).updateFunction(any(FunctionDetails.class), anyString());
+ verify(functions, times(1)).updateFunction(any(FunctionConfig.class), anyString());
}
setup();
[pulsar] 11/13: Fix unittests
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 625a31b952f32ed69753fd376e4f2a892effeba8
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Mon Oct 8 13:30:15 2018 -0700
Fix unittests
---
.../worker/PulsarWorkerAssignmentTest.java | 82 +++++++---------
.../apache/pulsar/io/PulsarFunctionTlsTest.java | 39 +++++++-
.../org/apache/pulsar/io/PulsarSinkE2ETest.java | 106 +++++++--------------
3 files changed, 105 insertions(+), 122 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
index 0f9219d..e3618ec 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
@@ -28,6 +28,7 @@ import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.URI;
+import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -51,6 +52,7 @@ import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.proto.Function.SinkSpec;
import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.sink.PulsarSink;
+import org.apache.pulsar.functions.utils.FunctionConfig;
import org.apache.pulsar.functions.utils.Reflections;
import org.apache.pulsar.functions.utils.Utils;
import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
@@ -192,13 +194,12 @@ public class PulsarWorkerAssignmentTest {
String jarFilePathUrl = Utils.FILE + ":"
+ PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
- FunctionDetails.Builder functionDetailsBuilder = createFunctionDetails(jarFilePathUrl, tenant, namespacePortion,
+ FunctionConfig functionConfig = createFunctionConfig(jarFilePathUrl, tenant, namespacePortion,
functionName, "my.*", sinkTopic, subscriptionName);
- functionDetailsBuilder.setParallelism(2);
- FunctionDetails functionDetails = functionDetailsBuilder.build();
+ functionConfig.setParallelism(2);
// (1) Create function with 2 instance
- admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+ admin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl);
retryStrategically((test) -> {
try {
return admin.topics().getStats(sinkTopic).subscriptions.size() == 1
@@ -213,10 +214,9 @@ public class PulsarWorkerAssignmentTest {
assertEquals(admin.topics().getStats(sinkTopic).subscriptions.values().iterator().next().consumers.size(), 2);
// (2) Update function with 1 instance
- functionDetailsBuilder.setParallelism(1);
- functionDetails = functionDetailsBuilder.build();
+ functionConfig.setParallelism(1);
// try to update function to test: update-function functionality
- admin.functions().updateFunctionWithUrl(functionDetails, jarFilePathUrl);
+ admin.functions().updateFunctionWithUrl(functionConfig, jarFilePathUrl);
retryStrategically((test) -> {
try {
return admin.topics().getStats(sinkTopic).subscriptions.size() == 1
@@ -247,17 +247,16 @@ public class PulsarWorkerAssignmentTest {
String jarFilePathUrl = Utils.FILE + ":"
+ PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
- FunctionDetails.Builder functionDetailsBuilder = null;
+ FunctionConfig functionConfig = null;
// (1) Register functions with 2 instances
for (int i = 0; i < totalFunctions; i++) {
String functionName = baseFunctionName + i;
- functionDetailsBuilder = createFunctionDetails(jarFilePathUrl, tenant, namespacePortion, functionName,
+ functionConfig = createFunctionConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
"my.*", sinkTopic, subscriptionName);
- functionDetailsBuilder.setParallelism(parallelism);
+ functionConfig.setParallelism(parallelism);
// set-auto-ack prop =true
- functionDetailsBuilder.setAutoAck(true);
- FunctionDetails functionDetails = functionDetailsBuilder.build();
- admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+ functionConfig.setAutoAck(true);
+ admin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl);
}
retryStrategically((test) -> {
try {
@@ -275,13 +274,12 @@ public class PulsarWorkerAssignmentTest {
// (2) Update function with prop=auto-ack and Delete 2 functions
for (int i = 0; i < totalFunctions; i++) {
String functionName = baseFunctionName + i;
- functionDetailsBuilder = createFunctionDetails(jarFilePathUrl, tenant, namespacePortion, functionName,
+ functionConfig = createFunctionConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
"my.*", sinkTopic, subscriptionName);
- functionDetailsBuilder.setParallelism(parallelism);
+ functionConfig.setParallelism(parallelism);
// set-auto-ack prop =false
- functionDetailsBuilder.setAutoAck(false);
- FunctionDetails functionDetails = functionDetailsBuilder.build();
- admin.functions().updateFunctionWithUrl(functionDetails, jarFilePathUrl);
+ functionConfig.setAutoAck(false);
+ admin.functions().updateFunctionWithUrl(functionConfig, jarFilePathUrl);
}
int totalDeletedFunction = 2;
@@ -328,8 +326,8 @@ public class PulsarWorkerAssignmentTest {
}
}
- protected static FunctionDetails.Builder createFunctionDetails(String jarFile, String tenant, String namespace,
- String functionName, String sourceTopic, String sinkTopic, String subscriptionName) {
+ protected static FunctionConfig createFunctionConfig(String jarFile, String tenant, String namespace,
+ String functionName, String sourceTopic, String sinkTopic, String subscriptionName) {
File file = new File(jarFile);
try {
@@ -340,35 +338,21 @@ public class PulsarWorkerAssignmentTest {
String sourceTopicPattern = String.format("persistent://%s/%s/%s", tenant, namespace, sourceTopic);
Class<?> typeArg = byte[].class;
- FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
- functionDetailsBuilder.setTenant(tenant);
- functionDetailsBuilder.setNamespace(namespace);
- functionDetailsBuilder.setName(functionName);
- functionDetailsBuilder.setRuntime(FunctionDetails.Runtime.JAVA);
- functionDetailsBuilder.setParallelism(1);
- functionDetailsBuilder.setClassName(IdentityFunction.class.getName());
-
- // set source spec
- // source spec classname should be empty so that the default pulsar source will be used
- SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
- sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.SHARED);
- sourceSpecBuilder.setTypeClassName(typeArg.getName());
- sourceSpecBuilder.setTopicsPattern(sourceTopicPattern);
- sourceSpecBuilder.setSubscriptionName(subscriptionName);
- sourceSpecBuilder.putTopicsToSerDeClassName(sourceTopicPattern, "");
- functionDetailsBuilder.setAutoAck(true);
- functionDetailsBuilder.setSource(sourceSpecBuilder);
-
- // set up sink spec
- SinkSpec.Builder sinkSpecBuilder = SinkSpec.newBuilder();
- // sinkSpecBuilder.setClassName(PulsarSink.class.getName());
- sinkSpecBuilder.setTopic(sinkTopic);
- Map<String, Object> sinkConfigMap = Maps.newHashMap();
- sinkSpecBuilder.setConfigs(new Gson().toJson(sinkConfigMap));
- sinkSpecBuilder.setTypeClassName(typeArg.getName());
- functionDetailsBuilder.setSink(sinkSpecBuilder);
-
- return functionDetailsBuilder;
+ FunctionConfig functionConfig = new FunctionConfig();
+ functionConfig.setTenant(tenant);
+ functionConfig.setNamespace(namespace);
+ functionConfig.setName(functionName);
+ functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
+ functionConfig.setParallelism(1);
+ functionConfig.setClassName(IdentityFunction.class.getName());
+
+ functionConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE);
+ functionConfig.setInputs(Collections.singleton(sourceTopicPattern));
+ functionConfig.setSubName(subscriptionName);
+ functionConfig.setAutoAck(true);
+ functionConfig.setOutput(sinkTopic);
+
+ return functionConfig;
}
}
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
index 2b32fd0..6d7dbd2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
@@ -25,13 +25,17 @@ import static org.mockito.Mockito.when;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import java.io.File;
import java.lang.reflect.Method;
+import java.net.MalformedURLException;
import java.net.URL;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import com.google.common.collect.Maps;
+import com.google.gson.Gson;
import org.apache.bookkeeper.test.PortManager;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.ServiceConfigurationUtils;
@@ -42,8 +46,12 @@ import org.apache.pulsar.client.admin.Tenants;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.functions.api.utils.IdentityFunction;
+import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.sink.PulsarSink;
+import org.apache.pulsar.functions.utils.FunctionConfig;
+import org.apache.pulsar.functions.utils.Reflections;
import org.apache.pulsar.functions.utils.Utils;
import org.apache.pulsar.functions.worker.FunctionMetaDataManager;
import org.apache.pulsar.functions.worker.WorkerConfig;
@@ -205,11 +213,11 @@ public class PulsarFunctionTlsTest {
String jarFilePathUrl = String.format("%s:%s", Utils.FILE,
PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath());
- FunctionDetails functionDetails = PulsarSinkE2ETest.createSinkConfig(jarFilePathUrl, tenant, namespacePortion,
+ FunctionConfig functionConfig = createFunctionConfig(jarFilePathUrl, tenant, namespacePortion,
functionName, "my.*", sinkTopic, subscriptionName);
try {
- functionAdmin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+ functionAdmin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl);
fail("Authentication should pass but call should fail with function already exist");
} catch (PulsarAdminException e) {
assertTrue(e.getMessage().contains("already exists"));
@@ -217,4 +225,31 @@ public class PulsarFunctionTlsTest {
}
+ protected static FunctionConfig createFunctionConfig(String jarFile, String tenant, String namespace, String functionName, String sourceTopic, String sinkTopic, String subscriptionName) {
+
+ File file = new File(jarFile);
+ try {
+ Reflections.loadJar(file);
+ } catch (MalformedURLException e) {
+ throw new RuntimeException("Failed to load user jar " + file, e);
+ }
+ String sourceTopicPattern = String.format("persistent://%s/%s/%s", tenant, namespace, sourceTopic);
+ Class<?> typeArg = byte[].class;
+
+ FunctionConfig functionConfig = new FunctionConfig();
+ functionConfig.setTenant(tenant);
+ functionConfig.setNamespace(namespace);
+ functionConfig.setName(functionName);
+ functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
+ functionConfig.setParallelism(1);
+ functionConfig.setClassName(IdentityFunction.class.getName());
+ functionConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE);
+ functionConfig.setSubName(subscriptionName);
+ functionConfig.setTopicsPattern(sourceTopicPattern);
+ functionConfig.setAutoAck(true);
+ functionConfig.setOutput(sinkTopic);
+
+ return functionConfig;
+ }
+
}
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
index 907cf86..1a7ec0f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
@@ -31,11 +31,7 @@ import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.URL;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.test.PortManager;
@@ -69,7 +65,9 @@ import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatusLis
import org.apache.pulsar.functions.proto.InstanceCommunication.MetricsData.DataDigest;
import org.apache.pulsar.functions.sink.PulsarSink;
import org.apache.pulsar.functions.source.TopicSchema;
+import org.apache.pulsar.functions.utils.FunctionConfig;
import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.functions.utils.SinkConfig;
import org.apache.pulsar.functions.utils.Utils;
import org.apache.pulsar.functions.worker.FunctionRuntimeManager;
import org.apache.pulsar.functions.worker.WorkerConfig;
@@ -270,12 +268,12 @@ public class PulsarSinkE2ETest {
String jarFilePathUrl = Utils.FILE + ":"
+ PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
- FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
+ SinkConfig sinkConfig = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
"my.*", sinkTopic, subscriptionName);
- admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+ admin.sink().createSinkWithUrl(sinkConfig, jarFilePathUrl);
// try to update function to test: update-function functionality
- admin.functions().updateFunctionWithUrl(functionDetails, jarFilePathUrl);
+ admin.sink().updateSinkWithUrl(sinkConfig, jarFilePathUrl);
retryStrategically((test) -> {
try {
@@ -332,12 +330,12 @@ public class PulsarSinkE2ETest {
String jarFilePathUrl = Utils.FILE + ":"
+ PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
- FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
+ SinkConfig sinkConfig = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
"my.*", sinkTopic, subscriptionName);
- admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+ admin.sink().createSinkWithUrl(sinkConfig, jarFilePathUrl);
// try to update function to test: update-function functionality
- admin.functions().updateFunctionWithUrl(functionDetails, jarFilePathUrl);
+ admin.sink().updateSinkWithUrl(sinkConfig, jarFilePathUrl);
retryStrategically((test) -> {
try {
@@ -382,7 +380,7 @@ public class PulsarSinkE2ETest {
assertEquals(ownerWorkerId, workerId);
}
- protected static FunctionDetails createSinkConfig(String jarFile, String tenant, String namespace, String functionName, String sourceTopic, String sinkTopic, String subscriptionName) {
+ protected static SinkConfig createSinkConfig(String jarFile, String tenant, String namespace, String functionName, String sourceTopic, String sinkTopic, String subscriptionName) {
File file = new File(jarFile);
try {
@@ -393,36 +391,17 @@ public class PulsarSinkE2ETest {
String sourceTopicPattern = String.format("persistent://%s/%s/%s", tenant, namespace, sourceTopic);
Class<?> typeArg = byte[].class;
- FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
- functionDetailsBuilder.setTenant(tenant);
- functionDetailsBuilder.setNamespace(namespace);
- functionDetailsBuilder.setName(functionName);
- functionDetailsBuilder.setRuntime(FunctionDetails.Runtime.JAVA);
- functionDetailsBuilder.setParallelism(1);
- functionDetailsBuilder.setClassName(IdentityFunction.class.getName());
- functionDetailsBuilder.setProcessingGuarantees(ProcessingGuarantees.EFFECTIVELY_ONCE);
-
- // set source spec
- // source spec classname should be empty so that the default pulsar source will be used
- SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
- sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.FAILOVER);
- sourceSpecBuilder.setTypeClassName(typeArg.getName());
- sourceSpecBuilder.setTopicsPattern(sourceTopicPattern);
- sourceSpecBuilder.setSubscriptionName(subscriptionName);
- sourceSpecBuilder.putTopicsToSerDeClassName(sourceTopicPattern, "");
- functionDetailsBuilder.setAutoAck(true);
- functionDetailsBuilder.setSource(sourceSpecBuilder);
-
- // set up sink spec
- SinkSpec.Builder sinkSpecBuilder = SinkSpec.newBuilder();
- // sinkSpecBuilder.setClassName(PulsarSink.class.getName());
- sinkSpecBuilder.setTopic(sinkTopic);
- Map<String, Object> sinkConfigMap = Maps.newHashMap();
- sinkSpecBuilder.setConfigs(new Gson().toJson(sinkConfigMap));
- sinkSpecBuilder.setTypeClassName(typeArg.getName());
- functionDetailsBuilder.setSink(sinkSpecBuilder);
-
- return functionDetailsBuilder.build();
+ SinkConfig sinkConfig = new SinkConfig();
+ sinkConfig.setTenant(tenant);
+ sinkConfig.setNamespace(namespace);
+ sinkConfig.setName(functionName);
+ sinkConfig.setParallelism(1);
+ sinkConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE);
+ sinkConfig.setSourceSubscriptionName(subscriptionName);
+ sinkConfig.setTopicsPattern(sourceTopicPattern);
+ sinkConfig.setAutoAck(true);
+ sinkConfig.setClassName(PulsarSink.class.getName());
+ return sinkConfig;
}
@Test(dataProvider = "validRoleName")
@@ -445,10 +424,10 @@ public class PulsarSinkE2ETest {
String jarFilePathUrl = Utils.FILE + ":"
+ PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
- FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
+ SinkConfig sinkConfig = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
"my.*", sinkTopic, subscriptionName);
try {
- admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+ admin.sink().createSinkWithUrl(sinkConfig, jarFilePathUrl);
assertTrue(validRoleName);
} catch (org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException ne) {
assertFalse(validRoleName);
@@ -473,35 +452,20 @@ public class PulsarSinkE2ETest {
String jarFilePathUrl = Utils.FILE + ":"
+ IdentityFunction.class.getProtectionDomain().getCodeSource().getLocation().getPath();
- FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
- functionDetailsBuilder.setTenant(tenant);
- functionDetailsBuilder.setNamespace(namespacePortion);
- functionDetailsBuilder.setName(functionName);
- functionDetailsBuilder.setRuntime(FunctionDetails.Runtime.JAVA);
- functionDetailsBuilder.setParallelism(1);
- functionDetailsBuilder.setClassName(IdentityFunction.class.getName());
+ SinkConfig sinkConfig = new SinkConfig();
+ sinkConfig.setTenant(tenant);
+ sinkConfig.setNamespace(namespacePortion);
+ sinkConfig.setName(functionName);
+ sinkConfig.setParallelism(1);
+ sinkConfig.setInputs(Collections.singleton(sinkTopic));
+ sinkConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE);
+ sinkConfig.setAutoAck(true);
Class<?>[] typeArgs = org.apache.pulsar.functions.utils.Utils.getFunctionTypes(new IdentityFunction(), false);
- // set source spec
- // source spec classname should be empty so that the default pulsar source will be used
- SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
- sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.FAILOVER);
- sourceSpecBuilder.putTopicsToSerDeClassName(sinkTopic, TopicSchema.DEFAULT_SERDE);
- functionDetailsBuilder.setAutoAck(true);
- functionDetailsBuilder.setSource(sourceSpecBuilder);
+ admin.sink().createSinkWithUrl(sinkConfig, jarFilePathUrl);
- // set up sink spec
- SinkSpec.Builder sinkSpecBuilder = SinkSpec.newBuilder();
- sinkSpecBuilder.setTopic(sinkTopic);
- Map<String, Object> sinkConfigMap = Maps.newHashMap();
- sinkSpecBuilder.setConfigs(new Gson().toJson(sinkConfigMap));
- functionDetailsBuilder.setSink(sinkSpecBuilder);
-
- FunctionDetails functionDetails = functionDetailsBuilder.build();
- admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
-
- FunctionDetails functionMetadata = admin.functions().getFunction(tenant, namespacePortion, functionName);
+ FunctionDetails functionMetadata = admin.source().getSource(tenant, namespacePortion, functionName);
assertEquals(functionMetadata.getSource().getTypeClassName(), typeArgs[0].getName());
assertEquals(functionMetadata.getSink().getTypeClassName(), typeArgs[1].getName());
@@ -527,9 +491,9 @@ public class PulsarSinkE2ETest {
String jarFilePathUrl = Utils.FILE + ":"
+ PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
- FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
+ SinkConfig sinkConfig = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
sourceTopicName, sinkTopic, subscriptionName);
- admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+ admin.sink().createSinkWithUrl(sinkConfig, jarFilePathUrl);
retryStrategically((test) -> {
try {
[pulsar] 10/13: Switched cli to use functionConfig instead of
function details
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 44de9f6e4977ff4a389d79e420cf5d7f26dd83d4
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Mon Oct 8 10:32:06 2018 -0700
Switched cli to use functionConfig instead of function details
---
.../org/apache/pulsar/client/admin/Functions.java | 19 ++++++------
.../client/admin/internal/FunctionsImpl.java | 34 ++++++++++++----------
.../org/apache/pulsar/admin/cli/CmdFunctions.java | 8 ++---
.../rest/api/v2/FunctionApiV2ResourceTest.java | 20 +++++++++----
4 files changed, 46 insertions(+), 35 deletions(-)
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 bdb2c56..10c890c 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
@@ -28,8 +28,7 @@ import org.apache.pulsar.common.io.ConnectorDefinition;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatusList;
-import org.apache.pulsar.functions.proto.InstanceCommunication.Metrics;
-import org.apache.pulsar.functions.worker.WorkerInfo;
+import org.apache.pulsar.functions.utils.FunctionConfig;
/**
* Admin interface for function management.
@@ -83,13 +82,13 @@ public interface Functions {
/**
* Create a new function.
*
- * @param functionDetails
+ * @param functionConfig
* the function configuration object
*
* @throws PulsarAdminException
* Unexpected error
*/
- void createFunction(FunctionDetails functionDetails, String fileName) throws PulsarAdminException;
+ void createFunction(FunctionConfig functionConfig, String fileName) throws PulsarAdminException;
/**
* <pre>
@@ -99,19 +98,19 @@ public interface Functions {
* Http: http://www.repo.com/fileName.jar
* </pre>
*
- * @param functionDetails
+ * @param functionConfig
* the function configuration object
* @param pkgUrl
* url from which pkg can be downloaded
* @throws PulsarAdminException
*/
- void createFunctionWithUrl(FunctionDetails functionDetails, String pkgUrl) throws PulsarAdminException;
+ void createFunctionWithUrl(FunctionConfig functionConfig, String pkgUrl) throws PulsarAdminException;
/**
* Update the configuration for a function.
* <p>
*
- * @param functionDetails
+ * @param functionConfig
* the function configuration object
*
* @throws NotAuthorizedException
@@ -121,7 +120,7 @@ public interface Functions {
* @throws PulsarAdminException
* Unexpected error
*/
- void updateFunction(FunctionDetails functionDetails, String fileName) throws PulsarAdminException;
+ void updateFunction(FunctionConfig functionConfig, String fileName) throws PulsarAdminException;
/**
* Update the configuration for a function.
@@ -132,7 +131,7 @@ public interface Functions {
* Http: http://www.repo.com/fileName.jar
* </pre>
*
- * @param functionDetails
+ * @param functionConfig
* the function configuration object
* @param pkgUrl
* url from which pkg can be downloaded
@@ -143,7 +142,7 @@ public interface Functions {
* @throws PulsarAdminException
* Unexpected error
*/
- void updateFunctionWithUrl(FunctionDetails functionDetails, String pkgUrl) throws PulsarAdminException;
+ void updateFunctionWithUrl(FunctionConfig functionConfig, String pkgUrl) throws PulsarAdminException;
/**
* Delete an existing 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 99d0da2..b9ea1a5 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
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.admin.internal;
+import com.google.gson.Gson;
import com.google.protobuf.AbstractMessage.Builder;
import com.google.protobuf.MessageOrBuilder;
import com.google.protobuf.util.JsonFormat;
@@ -48,6 +49,7 @@ import org.apache.pulsar.common.policies.data.ErrorData;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatusList;
+import org.apache.pulsar.functions.utils.FunctionConfig;
import org.apache.pulsar.functions.worker.WorkerInfo;
import org.glassfish.jersey.media.multipart.FormDataBodyPart;
import org.glassfish.jersey.media.multipart.FormDataMultiPart;
@@ -129,7 +131,7 @@ public class FunctionsImpl extends BaseResource implements Functions {
}
@Override
- public void createFunction(FunctionDetails functionDetails, String fileName) throws PulsarAdminException {
+ public void createFunction(FunctionConfig functionConfig, String fileName) throws PulsarAdminException {
try {
final FormDataMultiPart mp = new FormDataMultiPart();
@@ -138,10 +140,10 @@ public class FunctionsImpl extends BaseResource implements Functions {
mp.bodyPart(new FileDataBodyPart("data", new File(fileName), MediaType.APPLICATION_OCTET_STREAM_TYPE));
}
- mp.bodyPart(new FormDataBodyPart("functionDetails",
- printJson(functionDetails),
+ mp.bodyPart(new FormDataBodyPart("functionConfig",
+ new Gson().toJson(functionConfig),
MediaType.APPLICATION_JSON_TYPE));
- request(functions.path(functionDetails.getTenant()).path(functionDetails.getNamespace()).path(functionDetails.getName()))
+ request(functions.path(functionConfig.getTenant()).path(functionConfig.getNamespace()).path(functionConfig.getName()))
.post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
} catch (Exception e) {
throw getApiException(e);
@@ -149,16 +151,16 @@ public class FunctionsImpl extends BaseResource implements Functions {
}
@Override
- public void createFunctionWithUrl(FunctionDetails functionDetails, String pkgUrl) throws PulsarAdminException {
+ public void createFunctionWithUrl(FunctionConfig functionConfig, 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),
+ mp.bodyPart(new FormDataBodyPart("functionConfig",
+ new Gson().toJson(functionConfig),
MediaType.APPLICATION_JSON_TYPE));
- request(functions.path(functionDetails.getTenant()).path(functionDetails.getNamespace()).path(functionDetails.getName()))
+ request(functions.path(functionConfig.getTenant()).path(functionConfig.getNamespace()).path(functionConfig.getName()))
.post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
} catch (Exception e) {
throw getApiException(e);
@@ -176,7 +178,7 @@ public class FunctionsImpl extends BaseResource implements Functions {
}
@Override
- public void updateFunction(FunctionDetails functionDetails, String fileName) throws PulsarAdminException {
+ public void updateFunction(FunctionConfig functionConfig, String fileName) throws PulsarAdminException {
try {
final FormDataMultiPart mp = new FormDataMultiPart();
@@ -185,10 +187,10 @@ public class FunctionsImpl extends BaseResource implements Functions {
mp.bodyPart(new FileDataBodyPart("data", new File(fileName), MediaType.APPLICATION_OCTET_STREAM_TYPE));
}
- mp.bodyPart(new FormDataBodyPart("functionDetails",
- printJson(functionDetails),
+ mp.bodyPart(new FormDataBodyPart("functionConfig",
+ new Gson().toJson(functionConfig),
MediaType.APPLICATION_JSON_TYPE));
- request(functions.path(functionDetails.getTenant()).path(functionDetails.getNamespace()).path(functionDetails.getName()))
+ request(functions.path(functionConfig.getTenant()).path(functionConfig.getNamespace()).path(functionConfig.getName()))
.put(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
} catch (Exception e) {
throw getApiException(e);
@@ -196,16 +198,16 @@ public class FunctionsImpl extends BaseResource implements Functions {
}
@Override
- public void updateFunctionWithUrl(FunctionDetails functionDetails, String pkgUrl) throws PulsarAdminException {
+ public void updateFunctionWithUrl(FunctionConfig functionConfig, 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),
+ mp.bodyPart(new FormDataBodyPart("functionConfig", new Gson().toJson(functionConfig),
MediaType.APPLICATION_JSON_TYPE));
- request(functions.path(functionDetails.getTenant()).path(functionDetails.getNamespace())
- .path(functionDetails.getName())).put(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA),
+ request(functions.path(functionConfig.getTenant()).path(functionConfig.getNamespace())
+ .path(functionConfig.getName())).put(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA),
ErrorData.class);
} catch (Exception e) {
throw getApiException(e);
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index c771e7b..2f23a63 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -684,9 +684,9 @@ public class CmdFunctions extends CmdBase {
@Override
void runCmd() throws Exception {
if (Utils.isFunctionPackageUrlSupported(functionConfig.getJar())) {
- admin.functions().createFunctionWithUrl(FunctionConfigUtils.convert(functionConfig, classLoader), functionConfig.getJar());
+ admin.functions().createFunctionWithUrl(functionConfig, functionConfig.getJar());
} else {
- admin.functions().createFunction(FunctionConfigUtils.convert(functionConfig, classLoader), userCodeFile);
+ admin.functions().createFunction(functionConfig, userCodeFile);
}
print("Created successfully");
@@ -776,9 +776,9 @@ public class CmdFunctions extends CmdBase {
@Override
void runCmd() throws Exception {
if (Utils.isFunctionPackageUrlSupported(functionConfig.getJar())) {
- admin.functions().updateFunctionWithUrl(FunctionConfigUtils.convert(functionConfig, classLoader), functionConfig.getJar());
+ admin.functions().updateFunctionWithUrl(functionConfig, functionConfig.getJar());
} else {
- admin.functions().updateFunction(FunctionConfigUtils.convert(functionConfig, classLoader), userCodeFile);
+ admin.functions().updateFunction(functionConfig, userCodeFile);
}
print("Updated successfully");
}
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 9ce87d0..bd29941 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
@@ -346,6 +346,8 @@ public class FunctionApiV2ResourceTest {
null,
null,
new Gson().toJson(functionConfig),
+ null,
+ null,
null);
assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
@@ -372,7 +374,9 @@ public class FunctionApiV2ResourceTest {
null,
null,
new Gson().toJson(functionConfig),
- null);
+ null,
+ null,
+ null);
}
@Test
@@ -646,7 +650,9 @@ public class FunctionApiV2ResourceTest {
null,
null,
new Gson().toJson(functionConfig),
- null);
+ null,
+ null,
+ null);
assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
Assert.assertEquals(((ErrorData) response.getEntity()).reason, new ErrorData(expectedError).reason);
@@ -673,7 +679,9 @@ public class FunctionApiV2ResourceTest {
null,
null,
new Gson().toJson(functionConfig),
- null);
+ null,
+ null,
+ null);
}
@Test
@@ -756,7 +764,9 @@ public class FunctionApiV2ResourceTest {
filePackageUrl,
null,
new Gson().toJson(functionConfig),
- null);
+ null,
+ null,
+ null);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
}
@@ -1108,7 +1118,7 @@ public class FunctionApiV2ResourceTest {
functionConfig.setOutput(outputTopic);
functionConfig.setOutputSerdeClassName(outputSerdeClassName);
Response response = resource.registerFunction(tenant, namespace, function, null, null, filePackageUrl,
- null, new Gson().toJson(functionConfig), null);
+ null, new Gson().toJson(functionConfig), null, null, null);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
}
[pulsar] 09/13: Merge branch 'master' into
serverside_validation_endpoints
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d84aab19c768229d13a0a5faec3917b1c9e7e48b
Merge: 4334d62 7e3203d
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Mon Oct 8 09:33:16 2018 -0700
Merge branch 'master' into serverside_validation_endpoints
conf/broker.conf | 7 +
.../apache/pulsar/broker/ServiceConfiguration.java | 9 ++
.../apache/pulsar/broker/service/ServerCnx.java | 4 +-
.../apache/pulsar/broker/admin/AdminApiTest.java | 31 ++---
.../pulsar/broker/admin/v1/V1_AdminApiTest.java | 31 ++---
.../broker/auth/MockedPulsarServiceBaseTest.java | 1 +
.../pulsar/client/api/BrokerServiceLookupTest.java | 28 ++--
.../apache/pulsar/client/api/SimpleSchemaTest.java | 40 +++++-
.../java/org/apache/pulsar/client/api/Schema.java | 16 +++
.../client/impl/schema/AutoConsumeSchema.java | 7 +
.../client/impl/schema/AutoProduceBytesSchema.java | 24 +++-
.../pulsar/client/impl/schema/ByteSchema.java | 10 +-
.../pulsar/client/impl/schema/DoubleSchema.java | 11 +-
.../pulsar/client/impl/schema/FloatSchema.java | 11 +-
.../pulsar/client/impl/schema/IntSchema.java | 11 +-
.../pulsar/client/impl/schema/LongSchema.java | 11 +-
.../pulsar/client/impl/schema/ShortSchema.java | 11 +-
.../org/apache/pulsar/client/api/PulsarClient.java | 8 +-
.../instance/src/main/python/python_instance.py | 16 ++-
.../pulsar/functions/worker/SchedulerManager.java | 21 ++-
.../functions/worker/SchedulerManagerTest.java | 144 +++++++++++++++------
.../ProxyAuthenticatedProducerConsumerTest.java | 3 +-
.../proxy/server/ProxyAuthenticationTest.java | 7 +-
.../server/ProxyConnectionThrottlingTest.java | 5 +-
.../proxy/server/ProxyLookupThrottlingTest.java | 7 +-
.../org/apache/pulsar/proxy/server/ProxyTest.java | 9 +-
.../apache/pulsar/proxy/server/ProxyTlsTest.java | 5 +-
.../server/ProxyWithAuthorizationNegTest.java | 5 +-
.../proxy/server/ProxyWithAuthorizationTest.java | 3 +-
.../server/ProxyWithoutServiceDiscoveryTest.java | 3 +-
tests/integration/pom.xml | 6 -
.../integration/compat/kafka/KafkaApiTest.java | 1 +
32 files changed, 363 insertions(+), 143 deletions(-)
[pulsar] 13/13: Logic correction
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8fd2209494db0756ef83627da5a7426cc7e2af02
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Mon Oct 8 14:19:33 2018 -0700
Logic correction
---
.../apache/pulsar/functions/worker/rest/api/FunctionsImpl.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
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 08cd7de..04d865c 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
@@ -974,16 +974,16 @@ public class FunctionsImpl {
}
int numDefinitions = 0;
- if (StringUtils.isEmpty(functionDetailsJson)) {
+ if (!StringUtils.isEmpty(functionDetailsJson)) {
numDefinitions++;
}
- if (StringUtils.isEmpty(functionConfigJson)) {
+ if (!StringUtils.isEmpty(functionConfigJson)) {
numDefinitions++;
}
- if (StringUtils.isEmpty(sourceConfigJson)) {
+ if (!StringUtils.isEmpty(sourceConfigJson)) {
numDefinitions++;
}
- if (StringUtils.isEmpty(sinkConfigJson)) {
+ if (!StringUtils.isEmpty(sinkConfigJson)) {
numDefinitions++;
}
if (numDefinitions == 0) {
[pulsar] 08/13: Made pulsar admin to use Source/Sink endpoints
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4334d628fbffebc39d6d162911e1c9532c6fea83
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Mon Oct 8 09:32:34 2018 -0700
Made pulsar admin to use Source/Sink endpoints
---
pulsar-client-admin/pom.xml | 6 +
.../apache/pulsar/client/admin/PulsarAdmin.java | 36 +--
.../java/org/apache/pulsar/client/admin/Sink.java | 277 +++++++++++++++++++++
.../org/apache/pulsar/client/admin/Source.java | 277 +++++++++++++++++++++
.../pulsar/client/admin/internal/SinkImpl.java | 270 ++++++++++++++++++++
.../pulsar/client/admin/internal/SourceImpl.java | 270 ++++++++++++++++++++
.../java/org/apache/pulsar/admin/cli/CmdSinks.java | 24 +-
.../org/apache/pulsar/admin/cli/CmdSources.java | 24 +-
8 files changed, 1147 insertions(+), 37 deletions(-)
diff --git a/pulsar-client-admin/pom.xml b/pulsar-client-admin/pom.xml
index b42ea2c..ac39524 100644
--- a/pulsar-client-admin/pom.xml
+++ b/pulsar-client-admin/pom.xml
@@ -46,6 +46,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-functions-utils</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.glassfish.jersey.core</groupId>
<artifactId>jersey-client</artifactId>
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java
index a1502b0..0953d22 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java
@@ -32,21 +32,7 @@ import javax.ws.rs.client.WebTarget;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.conn.ssl.DefaultHostnameVerifier;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
-import org.apache.pulsar.client.admin.internal.BookiesImpl;
-import org.apache.pulsar.client.admin.internal.BrokerStatsImpl;
-import org.apache.pulsar.client.admin.internal.BrokersImpl;
-import org.apache.pulsar.client.admin.internal.ClustersImpl;
-import org.apache.pulsar.client.admin.internal.FunctionsImpl;
-import org.apache.pulsar.client.admin.internal.JacksonConfigurator;
-import org.apache.pulsar.client.admin.internal.LookupImpl;
-import org.apache.pulsar.client.admin.internal.NamespacesImpl;
-import org.apache.pulsar.client.admin.internal.NonPersistentTopicsImpl;
-import org.apache.pulsar.client.admin.internal.SchemasImpl;
-import org.apache.pulsar.client.admin.internal.TopicsImpl;
-import org.apache.pulsar.client.admin.internal.WorkerImpl;
-import org.apache.pulsar.client.admin.internal.TenantsImpl;
-import org.apache.pulsar.client.admin.internal.PulsarAdminBuilderImpl;
-import org.apache.pulsar.client.admin.internal.ResourceQuotasImpl;
+import org.apache.pulsar.client.admin.internal.*;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.client.api.AuthenticationFactory;
@@ -85,6 +71,8 @@ public class PulsarAdmin implements Closeable {
private final String serviceUrl;
private final Lookup lookups;
private final Functions functions;
+ private final Source source;
+ private final Sink sink;
private final Worker worker;
private final Schemas schemas;
protected final WebTarget root;
@@ -189,6 +177,8 @@ public class PulsarAdmin implements Closeable {
this.resourceQuotas = new ResourceQuotasImpl(root, auth);
this.lookups = new LookupImpl(root, auth, useTls);
this.functions = new FunctionsImpl(root, auth);
+ this.source = new SourceImpl(root, auth);
+ this.sink = new SinkImpl(root, auth);
this.worker = new WorkerImpl(root, auth);
this.schemas = new SchemasImpl(root, auth);
this.bookies = new BookiesImpl(root, auth);
@@ -358,6 +348,22 @@ public class PulsarAdmin implements Closeable {
}
/**
+ *
+ * @return the source management object
+ */
+ public Source source() {
+ return source;
+ }
+
+ /**
+ *
+ * @return the sink management object
+ */
+ public Sink sink() {
+ return sink;
+ }
+
+ /**
*
* @return the Worker stats
*/
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java
new file mode 100644
index 0000000..3f8fe2f
--- /dev/null
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java
@@ -0,0 +1,277 @@
+/**
+ * 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.client.admin;
+
+import org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException;
+import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
+import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatusList;
+import org.apache.pulsar.functions.utils.SinkConfig;
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Admin interface for Sink management.
+ */
+public interface Sink {
+ /**
+ * Get the list of sinks.
+ * <p>
+ * Get the list of all the Pulsar Sinks.
+ * <p>
+ * Response Example:
+ *
+ * <pre>
+ * <code>["f1", "f2", "f3"]</code>
+ * </pre>
+ *
+ * @throws NotAuthorizedException
+ * Don't have admin permission
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ List<String> getSinks(String tenant, String namespace) throws PulsarAdminException;
+
+ /**
+ * Get the configuration for the specified sink.
+ * <p>
+ * Response Example:
+ *
+ * <pre>
+ * <code>{ serviceUrl : "http://my-broker.example.com:8080/" }</code>
+ * </pre>
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param sink
+ * Sink name
+ *
+ * @return the sink configuration
+ *
+ * @throws NotAuthorizedException
+ * You don't have admin permission to get the configuration of the cluster
+ * @throws NotFoundException
+ * Cluster doesn't exist
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ Function.FunctionDetails getSink(String tenant, String namespace, String sink) throws PulsarAdminException;
+
+ /**
+ * Create a new sink.
+ *
+ * @param sinkConfig
+ * the sink configuration object
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void createSink(SinkConfig sinkConfig, String fileName) throws PulsarAdminException;
+
+ /**
+ * <pre>
+ * Create a new sink 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 sinkConfig
+ * the sink configuration object
+ * @param pkgUrl
+ * url from which pkg can be downloaded
+ * @throws PulsarAdminException
+ */
+ void createSinkWithUrl(SinkConfig sinkConfig, String pkgUrl) throws PulsarAdminException;
+
+ /**
+ * Update the configuration for a sink.
+ * <p>
+ *
+ * @param sinkConfig
+ * the sink configuration object
+ *
+ * @throws NotAuthorizedException
+ * You don't have admin permission to create the cluster
+ * @throws NotFoundException
+ * Cluster doesn't exist
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void updateSink(SinkConfig sinkConfig, String fileName) throws PulsarAdminException;
+
+ /**
+ * Update the configuration for a sink.
+ * <pre>
+ * Update a sink 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 sinkConfig
+ * the sink configuration object
+ * @param pkgUrl
+ * url from which pkg can be downloaded
+ * @throws NotAuthorizedException
+ * You don't have admin permission to create the cluster
+ * @throws NotFoundException
+ * Cluster doesn't exist
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void updateSinkWithUrl(SinkConfig sinkConfig, String pkgUrl) throws PulsarAdminException;
+
+ /**
+ * Delete an existing sink
+ * <p>
+ * Delete a sink
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param sink
+ * Sink name
+ *
+ * @throws NotAuthorizedException
+ * You don't have admin permission
+ * @throws NotFoundException
+ * Cluster does not exist
+ * @throws PreconditionFailedException
+ * Cluster is not empty
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void deleteSink(String tenant, String namespace, String sink) throws PulsarAdminException;
+
+ /**
+ * Gets the current status of a sink.
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param sink
+ * Sink name
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ FunctionStatusList getSinkStatus(String tenant, String namespace, String sink) throws PulsarAdminException;
+
+ /**
+ * Gets the current status of a sink instance.
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param sink
+ * Sink name
+ * @param id
+ * Sink instance-id
+ * @return
+ * @throws PulsarAdminException
+ */
+ FunctionStatus getSinkStatus(String tenant, String namespace, String sink, int id)
+ throws PulsarAdminException;
+
+ /**
+ * Restart sink instance
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param sink
+ * Sink name
+ *
+ * @param instanceId
+ * Sink instanceId
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void restartSink(String tenant, String namespace, String sink, int instanceId) throws PulsarAdminException;
+
+ /**
+ * Restart all sink instances
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param sink
+ * Sink name
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void restartSink(String tenant, String namespace, String sink) throws PulsarAdminException;
+
+
+ /**
+ * Stop sink instance
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param sink
+ * Sink name
+ *
+ * @param instanceId
+ * Sink instanceId
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void stopSink(String tenant, String namespace, String sink, int instanceId) throws PulsarAdminException;
+
+ /**
+ * Stop all sink instances
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param sink
+ * Sink name
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void stopSink(String tenant, String namespace, String sink) throws PulsarAdminException;
+
+ /**
+ * Fetches a list of supported Pulsar IO sinks currently running in cluster mode
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ *
+ */
+ List<ConnectorDefinition> getBuiltInSinks() throws PulsarAdminException;
+}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java
new file mode 100644
index 0000000..3c43cf2
--- /dev/null
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java
@@ -0,0 +1,277 @@
+/**
+ * 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.client.admin;
+
+import org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException;
+import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
+import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatusList;
+import org.apache.pulsar.functions.utils.SourceConfig;
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Admin interface for Source management.
+ */
+public interface Source {
+ /**
+ * Get the list of sources.
+ * <p>
+ * Get the list of all the Pulsar Sources.
+ * <p>
+ * Response Example:
+ *
+ * <pre>
+ * <code>["f1", "f2", "f3"]</code>
+ * </pre>
+ *
+ * @throws NotAuthorizedException
+ * Don't have admin permission
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ List<String> getSources(String tenant, String namespace) throws PulsarAdminException;
+
+ /**
+ * Get the configuration for the specified source.
+ * <p>
+ * Response Example:
+ *
+ * <pre>
+ * <code>{ serviceUrl : "http://my-broker.example.com:8080/" }</code>
+ * </pre>
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param source
+ * Source name
+ *
+ * @return the source configuration
+ *
+ * @throws NotAuthorizedException
+ * You don't have admin permission to get the configuration of the cluster
+ * @throws NotFoundException
+ * Cluster doesn't exist
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ Function.FunctionDetails getSource(String tenant, String namespace, String source) throws PulsarAdminException;
+
+ /**
+ * Create a new source.
+ *
+ * @param sourceConfig
+ * the source configuration object
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void createSource(SourceConfig sourceConfig, String fileName) throws PulsarAdminException;
+
+ /**
+ * <pre>
+ * Create a new source 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 sourceConfig
+ * the source configuration object
+ * @param pkgUrl
+ * url from which pkg can be downloaded
+ * @throws PulsarAdminException
+ */
+ void createSourceWithUrl(SourceConfig sourceConfig, String pkgUrl) throws PulsarAdminException;
+
+ /**
+ * Update the configuration for a source.
+ * <p>
+ *
+ * @param sourceConfig
+ * the source configuration object
+ *
+ * @throws NotAuthorizedException
+ * You don't have admin permission to create the cluster
+ * @throws NotFoundException
+ * Cluster doesn't exist
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void updateSource(SourceConfig sourceConfig, String fileName) throws PulsarAdminException;
+
+ /**
+ * Update the configuration for a source.
+ * <pre>
+ * Update a source 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 sourceConfig
+ * the source configuration object
+ * @param pkgUrl
+ * url from which pkg can be downloaded
+ * @throws NotAuthorizedException
+ * You don't have admin permission to create the cluster
+ * @throws NotFoundException
+ * Cluster doesn't exist
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void updateSourceWithUrl(SourceConfig sourceConfig, String pkgUrl) throws PulsarAdminException;
+
+ /**
+ * Delete an existing source
+ * <p>
+ * Delete a source
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param source
+ * Source name
+ *
+ * @throws NotAuthorizedException
+ * You don't have admin permission
+ * @throws NotFoundException
+ * Cluster does not exist
+ * @throws PreconditionFailedException
+ * Cluster is not empty
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void deleteSource(String tenant, String namespace, String source) throws PulsarAdminException;
+
+ /**
+ * Gets the current status of a source.
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param source
+ * Source name
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ FunctionStatusList getSourceStatus(String tenant, String namespace, String source) throws PulsarAdminException;
+
+ /**
+ * Gets the current status of a source instance.
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param source
+ * Source name
+ * @param id
+ * Source instance-id
+ * @return
+ * @throws PulsarAdminException
+ */
+ FunctionStatus getSourceStatus(String tenant, String namespace, String source, int id)
+ throws PulsarAdminException;
+
+ /**
+ * Restart source instance
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param source
+ * Source name
+ *
+ * @param instanceId
+ * Source instanceId
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void restartSource(String tenant, String namespace, String source, int instanceId) throws PulsarAdminException;
+
+ /**
+ * Restart all source instances
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param source
+ * Source name
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void restartSource(String tenant, String namespace, String source) throws PulsarAdminException;
+
+
+ /**
+ * Stop source instance
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param source
+ * Source name
+ *
+ * @param instanceId
+ * Source instanceId
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void stopSource(String tenant, String namespace, String source, int instanceId) throws PulsarAdminException;
+
+ /**
+ * Stop all source instances
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param source
+ * Source name
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void stopSource(String tenant, String namespace, String source) throws PulsarAdminException;
+
+ /**
+ * Fetches a list of supported Pulsar IO sources currently running in cluster mode
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ *
+ */
+ List<ConnectorDefinition> getBuiltInSources() throws PulsarAdminException;
+}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinkImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinkImpl.java
new file mode 100644
index 0000000..b74f4ff
--- /dev/null
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinkImpl.java
@@ -0,0 +1,270 @@
+/**
+ * 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.client.admin.internal;
+
+import com.google.gson.Gson;
+import com.google.protobuf.AbstractMessage.Builder;
+import com.google.protobuf.util.JsonFormat;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.Sink;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.common.policies.data.ErrorData;
+import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatusList;
+import org.apache.pulsar.functions.utils.SinkConfig;
+import org.glassfish.jersey.media.multipart.FormDataBodyPart;
+import org.glassfish.jersey.media.multipart.FormDataMultiPart;
+import org.glassfish.jersey.media.multipart.file.FileDataBodyPart;
+
+import javax.ws.rs.ClientErrorException;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.GenericType;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class SinkImpl extends BaseResource implements Sink {
+
+ private final WebTarget sink;
+
+ public SinkImpl(WebTarget web, Authentication auth) {
+ super(auth);
+ this.sink = web.path("/admin/sink");
+ }
+
+ @Override
+ public List<String> getSinks(String tenant, String namespace) throws PulsarAdminException {
+ try {
+ Response response = request(sink.path(tenant).path(namespace)).get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ return response.readEntity(new GenericType<List<String>>() {
+ });
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public FunctionDetails getSink(String tenant, String namespace, String sinkName) throws PulsarAdminException {
+ try {
+ Response response = request(sink.path(tenant).path(namespace).path(sinkName)).get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ String jsonResponse = response.readEntity(String.class);
+ FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
+ mergeJson(jsonResponse, functionDetailsBuilder);
+ return functionDetailsBuilder.build();
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public FunctionStatusList getSinkStatus(
+ String tenant, String namespace, String sinkName) throws PulsarAdminException {
+ try {
+ Response response = request(sink.path(tenant).path(namespace).path(sinkName).path("status")).get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ String jsonResponse = response.readEntity(String.class);
+ FunctionStatusList.Builder functionStatusBuilder = FunctionStatusList.newBuilder();
+ mergeJson(jsonResponse, functionStatusBuilder);
+ return functionStatusBuilder.build();
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public FunctionStatus getSinkStatus(
+ String tenant, String namespace, String sinkName, int id) throws PulsarAdminException {
+ try {
+ Response response = request(
+ sink.path(tenant).path(namespace).path(sinkName).path(Integer.toString(id)).path("status"))
+ .get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ String jsonResponse = response.readEntity(String.class);
+ FunctionStatus.Builder functionStatusBuilder = FunctionStatus.newBuilder();
+ mergeJson(jsonResponse, functionStatusBuilder);
+ return functionStatusBuilder.build();
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void createSink(SinkConfig sinkConfig, String fileName) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ if (fileName != null && !fileName.startsWith("builtin://")) {
+ // If the function code is built in, we don't need to submit here
+ mp.bodyPart(new FileDataBodyPart("data", new File(fileName), MediaType.APPLICATION_OCTET_STREAM_TYPE));
+ }
+
+ mp.bodyPart(new FormDataBodyPart("sinkConfig",
+ new Gson().toJson(sinkConfig),
+ MediaType.APPLICATION_JSON_TYPE));
+ request(sink.path(sinkConfig.getTenant()).path(sinkConfig.getNamespace()).path(sinkConfig.getName()))
+ .post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void createSinkWithUrl(SinkConfig sinkConfig, String pkgUrl) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ mp.bodyPart(new FormDataBodyPart("url", pkgUrl, MediaType.TEXT_PLAIN_TYPE));
+
+ mp.bodyPart(new FormDataBodyPart("sinkConfig",
+ new Gson().toJson(sinkConfig),
+ MediaType.APPLICATION_JSON_TYPE));
+ request(sink.path(sinkConfig.getTenant()).path(sinkConfig.getNamespace()).path(sinkConfig.getName()))
+ .post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void deleteSink(String cluster, String namespace, String function) throws PulsarAdminException {
+ try {
+ request(sink.path(cluster).path(namespace).path(function))
+ .delete(ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void updateSink(SinkConfig sinkConfig, String fileName) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ if (fileName != null && !fileName.startsWith("builtin://")) {
+ // If the function code is built in, we don't need to submit here
+ mp.bodyPart(new FileDataBodyPart("data", new File(fileName), MediaType.APPLICATION_OCTET_STREAM_TYPE));
+ }
+
+ mp.bodyPart(new FormDataBodyPart("sinkConfig",
+ new Gson().toJson(sinkConfig),
+ MediaType.APPLICATION_JSON_TYPE));
+ request(sink.path(sinkConfig.getTenant()).path(sinkConfig.getNamespace()).path(sinkConfig.getName()))
+ .put(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void updateSinkWithUrl(SinkConfig sinkConfig, String pkgUrl) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ mp.bodyPart(new FormDataBodyPart("url", pkgUrl, MediaType.TEXT_PLAIN_TYPE));
+
+ mp.bodyPart(new FormDataBodyPart("sinkConfig", new Gson().toJson(sinkConfig),
+ MediaType.APPLICATION_JSON_TYPE));
+ request(sink.path(sinkConfig.getTenant()).path(sinkConfig.getNamespace())
+ .path(sinkConfig.getName())).put(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA),
+ ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void restartSink(String tenant, String namespace, String functionName, int instanceId)
+ throws PulsarAdminException {
+ try {
+ request(sink.path(tenant).path(namespace).path(functionName).path(Integer.toString(instanceId))
+ .path("restart")).post(Entity.entity("", MediaType.APPLICATION_JSON), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void restartSink(String tenant, String namespace, String functionName) throws PulsarAdminException {
+ try {
+ request(sink.path(tenant).path(namespace).path(functionName).path("restart"))
+ .post(Entity.entity("", MediaType.APPLICATION_JSON), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void stopSink(String tenant, String namespace, String sinkName, int instanceId)
+ throws PulsarAdminException {
+ try {
+ request(sink.path(tenant).path(namespace).path(sinkName).path(Integer.toString(instanceId))
+ .path("stop")).post(Entity.entity("", MediaType.APPLICATION_JSON), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void stopSink(String tenant, String namespace, String sinkName) throws PulsarAdminException {
+ try {
+ request(sink.path(tenant).path(namespace).path(sinkName).path("stop"))
+ .post(Entity.entity("", MediaType.APPLICATION_JSON), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public List<ConnectorDefinition> getBuiltInSinks() throws PulsarAdminException {
+ try {
+ Response response = request(sink.path("builtinsink")).get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ return response.readEntity(new GenericType<List<ConnectorDefinition>>() {});
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+
+ public static void mergeJson(String json, Builder builder) throws IOException {
+ JsonFormat.parser().merge(json, builder);
+ }
+
+}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourceImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourceImpl.java
new file mode 100644
index 0000000..3ae0a6f
--- /dev/null
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourceImpl.java
@@ -0,0 +1,270 @@
+/**
+ * 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.client.admin.internal;
+
+import com.google.gson.Gson;
+import com.google.protobuf.AbstractMessage.Builder;
+import com.google.protobuf.util.JsonFormat;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.Source;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.common.policies.data.ErrorData;
+import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatusList;
+import org.apache.pulsar.functions.utils.SourceConfig;
+import org.glassfish.jersey.media.multipart.FormDataBodyPart;
+import org.glassfish.jersey.media.multipart.FormDataMultiPart;
+import org.glassfish.jersey.media.multipart.file.FileDataBodyPart;
+
+import javax.ws.rs.ClientErrorException;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.GenericType;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class SourceImpl extends BaseResource implements Source {
+
+ private final WebTarget source;
+
+ public SourceImpl(WebTarget web, Authentication auth) {
+ super(auth);
+ this.source = web.path("/admin/source");
+ }
+
+ @Override
+ public List<String> getSources(String tenant, String namespace) throws PulsarAdminException {
+ try {
+ Response response = request(source.path(tenant).path(namespace)).get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ return response.readEntity(new GenericType<List<String>>() {
+ });
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public FunctionDetails getSource(String tenant, String namespace, String sourceName) throws PulsarAdminException {
+ try {
+ Response response = request(source.path(tenant).path(namespace).path(sourceName)).get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ String jsonResponse = response.readEntity(String.class);
+ FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
+ mergeJson(jsonResponse, functionDetailsBuilder);
+ return functionDetailsBuilder.build();
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public FunctionStatusList getSourceStatus(
+ String tenant, String namespace, String sourceName) throws PulsarAdminException {
+ try {
+ Response response = request(source.path(tenant).path(namespace).path(sourceName).path("status")).get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ String jsonResponse = response.readEntity(String.class);
+ FunctionStatusList.Builder functionStatusBuilder = FunctionStatusList.newBuilder();
+ mergeJson(jsonResponse, functionStatusBuilder);
+ return functionStatusBuilder.build();
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public FunctionStatus getSourceStatus(
+ String tenant, String namespace, String sourceName, int id) throws PulsarAdminException {
+ try {
+ Response response = request(
+ source.path(tenant).path(namespace).path(sourceName).path(Integer.toString(id)).path("status"))
+ .get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ String jsonResponse = response.readEntity(String.class);
+ FunctionStatus.Builder functionStatusBuilder = FunctionStatus.newBuilder();
+ mergeJson(jsonResponse, functionStatusBuilder);
+ return functionStatusBuilder.build();
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void createSource(SourceConfig sourceConfig, String fileName) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ if (fileName != null && !fileName.startsWith("builtin://")) {
+ // If the function code is built in, we don't need to submit here
+ mp.bodyPart(new FileDataBodyPart("data", new File(fileName), MediaType.APPLICATION_OCTET_STREAM_TYPE));
+ }
+
+ mp.bodyPart(new FormDataBodyPart("sourceConfig",
+ new Gson().toJson(sourceConfig),
+ MediaType.APPLICATION_JSON_TYPE));
+ request(source.path(sourceConfig.getTenant()).path(sourceConfig.getNamespace()).path(sourceConfig.getName()))
+ .post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void createSourceWithUrl(SourceConfig sourceConfig, String pkgUrl) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ mp.bodyPart(new FormDataBodyPart("url", pkgUrl, MediaType.TEXT_PLAIN_TYPE));
+
+ mp.bodyPart(new FormDataBodyPart("sourceConfig",
+ new Gson().toJson(sourceConfig),
+ MediaType.APPLICATION_JSON_TYPE));
+ request(source.path(sourceConfig.getTenant()).path(sourceConfig.getNamespace()).path(sourceConfig.getName()))
+ .post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void deleteSource(String cluster, String namespace, String function) throws PulsarAdminException {
+ try {
+ request(source.path(cluster).path(namespace).path(function))
+ .delete(ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void updateSource(SourceConfig sourceConfig, String fileName) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ if (fileName != null && !fileName.startsWith("builtin://")) {
+ // If the function code is built in, we don't need to submit here
+ mp.bodyPart(new FileDataBodyPart("data", new File(fileName), MediaType.APPLICATION_OCTET_STREAM_TYPE));
+ }
+
+ mp.bodyPart(new FormDataBodyPart("sourceConfig",
+ new Gson().toJson(sourceConfig),
+ MediaType.APPLICATION_JSON_TYPE));
+ request(source.path(sourceConfig.getTenant()).path(sourceConfig.getNamespace()).path(sourceConfig.getName()))
+ .put(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void updateSourceWithUrl(SourceConfig sourceConfig, String pkgUrl) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ mp.bodyPart(new FormDataBodyPart("url", pkgUrl, MediaType.TEXT_PLAIN_TYPE));
+
+ mp.bodyPart(new FormDataBodyPart("sourceConfig", new Gson().toJson(sourceConfig),
+ MediaType.APPLICATION_JSON_TYPE));
+ request(source.path(sourceConfig.getTenant()).path(sourceConfig.getNamespace())
+ .path(sourceConfig.getName())).put(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA),
+ ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void restartSource(String tenant, String namespace, String functionName, int instanceId)
+ throws PulsarAdminException {
+ try {
+ request(source.path(tenant).path(namespace).path(functionName).path(Integer.toString(instanceId))
+ .path("restart")).post(Entity.entity("", MediaType.APPLICATION_JSON), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void restartSource(String tenant, String namespace, String functionName) throws PulsarAdminException {
+ try {
+ request(source.path(tenant).path(namespace).path(functionName).path("restart"))
+ .post(Entity.entity("", MediaType.APPLICATION_JSON), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void stopSource(String tenant, String namespace, String sourceName, int instanceId)
+ throws PulsarAdminException {
+ try {
+ request(source.path(tenant).path(namespace).path(sourceName).path(Integer.toString(instanceId))
+ .path("stop")).post(Entity.entity("", MediaType.APPLICATION_JSON), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void stopSource(String tenant, String namespace, String sourceName) throws PulsarAdminException {
+ try {
+ request(source.path(tenant).path(namespace).path(sourceName).path("stop"))
+ .post(Entity.entity("", MediaType.APPLICATION_JSON), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public List<ConnectorDefinition> getBuiltInSources() throws PulsarAdminException {
+ try {
+ Response response = request(source.path("builtinsource")).get();
+ if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ throw new ClientErrorException(response);
+ }
+ return response.readEntity(new GenericType<List<ConnectorDefinition>>() {});
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+
+ public static void mergeJson(String json, Builder builder) throws IOException {
+ JsonFormat.parser().merge(json, builder);
+ }
+
+}
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
index bd5af79..e56a343 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
@@ -40,6 +40,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -79,7 +80,7 @@ public class CmdSinks extends CmdBase {
jcommander.addCommand("update", updateSink);
jcommander.addCommand("delete", deleteSink);
jcommander.addCommand("localrun", localSinkRunner);
- jcommander.addCommand("available-sinks", new ListSinks());
+ jcommander.addCommand("available-sinks", new ListBuiltInSinks());
}
/**
@@ -187,9 +188,9 @@ public class CmdSinks extends CmdBase {
@Override
void runCmd() throws Exception {
if (Utils.isFunctionPackageUrlSupported(archive)) {
- admin.functions().createFunctionWithUrl(SinkConfigUtils.convert(sinkConfig), sinkConfig.getArchive());
+ admin.sink().createSinkWithUrl(sinkConfig, sinkConfig.getArchive());
} else {
- admin.functions().createFunction(SinkConfigUtils.convert(sinkConfig), sinkConfig.getArchive());
+ admin.sink().createSink(sinkConfig, sinkConfig.getArchive());
}
print("Created successfully");
}
@@ -200,9 +201,9 @@ public class CmdSinks extends CmdBase {
@Override
void runCmd() throws Exception {
if (Utils.isFunctionPackageUrlSupported(archive)) {
- admin.functions().updateFunctionWithUrl(SinkConfigUtils.convert(sinkConfig), sinkConfig.getArchive());
+ admin.sink().updateSinkWithUrl(sinkConfig, sinkConfig.getArchive());
} else {
- admin.functions().updateFunction(SinkConfigUtils.convert(sinkConfig), sinkConfig.getArchive());
+ admin.sink().updateSink(sinkConfig, sinkConfig.getArchive());
}
print("Updated successfully");
}
@@ -282,6 +283,8 @@ public class CmdSinks extends CmdBase {
protected SinkConfig sinkConfig;
+ protected NarClassLoader classLoader;
+
private void mergeArgs() {
if (!StringUtils.isBlank(DEPRECATED_subsName)) subsName = DEPRECATED_subsName;
if (!StringUtils.isBlank(DEPRECATED_topicsPattern)) topicsPattern = DEPRECATED_topicsPattern;
@@ -449,7 +452,6 @@ public class CmdSinks extends CmdBase {
}
// if jar file is present locally then load jar and validate SinkClass in it
- ClassLoader classLoader = null;
if (archivePath != null) {
if (!fileExists(archivePath)) {
throw new ParameterException("Archive file " + archivePath + " does not exist");
@@ -482,14 +484,14 @@ public class CmdSinks extends CmdBase {
throws IOException {
org.apache.pulsar.functions.proto.Function.FunctionDetails.Builder functionDetailsBuilder
= org.apache.pulsar.functions.proto.Function.FunctionDetails.newBuilder();
- Utils.mergeJson(FunctionsImpl.printJson(SinkConfigUtils.convert(sinkConfig)), functionDetailsBuilder);
+ Utils.mergeJson(FunctionsImpl.printJson(SinkConfigUtils.convert(sinkConfig, classLoader)), functionDetailsBuilder);
return functionDetailsBuilder.build();
}
protected String validateSinkType(String sinkType) throws IOException {
Set<String> availableSinks;
try {
- availableSinks = admin.functions().getSinks();
+ availableSinks = admin.sink().getBuiltInSinks().stream().map(ConnectorDefinition::getName).collect(Collectors.toSet());
} catch (PulsarAdminException e) {
throw new IOException(e);
}
@@ -533,16 +535,16 @@ public class CmdSinks extends CmdBase {
@Override
void runCmd() throws Exception {
- admin.functions().deleteFunction(tenant, namespace, name);
+ admin.sink().deleteSink(tenant, namespace, name);
print("Deleted successfully");
}
}
@Parameters(commandDescription = "Get the list of Pulsar IO connector sinks supported by Pulsar cluster")
- public class ListSinks extends BaseCommand {
+ public class ListBuiltInSinks extends BaseCommand {
@Override
void runCmd() throws Exception {
- admin.functions().getConnectorsList().stream().filter(x -> isNotBlank(x.getSinkClass()))
+ admin.sink().getBuiltInSinks().stream().filter(x -> isNotBlank(x.getSinkClass()))
.forEach(connector -> {
System.out.println(connector.getName());
System.out.println(WordUtils.wrap(connector.getDescription(), 80));
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
index 5d0c84c..f2d7681 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
@@ -37,6 +37,7 @@ import java.nio.file.Paths;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -79,7 +80,7 @@ public class CmdSources extends CmdBase {
jcommander.addCommand("update", updateSource);
jcommander.addCommand("delete", deleteSource);
jcommander.addCommand("localrun", localSourceRunner);
- jcommander.addCommand("available-sources", new ListSources());
+ jcommander.addCommand("available-sources", new ListBuiltInSources());
}
/**
@@ -187,9 +188,9 @@ public class CmdSources extends CmdBase {
@Override
void runCmd() throws Exception {
if (Utils.isFunctionPackageUrlSupported(this.sourceConfig.getArchive())) {
- admin.functions().createFunctionWithUrl(SourceConfigUtils.convert(sourceConfig), sourceConfig.getArchive());
+ admin.source().createSourceWithUrl(sourceConfig, sourceConfig.getArchive());
} else {
- admin.functions().createFunction(SourceConfigUtils.convert(sourceConfig), sourceConfig.getArchive());
+ admin.source().createSource(sourceConfig, sourceConfig.getArchive());
}
print("Created successfully");
}
@@ -200,9 +201,9 @@ public class CmdSources extends CmdBase {
@Override
void runCmd() throws Exception {
if (Utils.isFunctionPackageUrlSupported(sourceConfig.getArchive())) {
- admin.functions().updateFunctionWithUrl(SourceConfigUtils.convert(sourceConfig), sourceConfig.getArchive());
+ admin.source().updateSourceWithUrl(sourceConfig, sourceConfig.getArchive());
} else {
- admin.functions().updateFunction(SourceConfigUtils.convert(sourceConfig), sourceConfig.getArchive());
+ admin.source().updateSource(sourceConfig, sourceConfig.getArchive());
}
print("Updated successfully");
}
@@ -267,6 +268,8 @@ public class CmdSources extends CmdBase {
protected SourceConfig sourceConfig;
+ protected NarClassLoader classLoader;
+
private void mergeArgs() {
if (DEPRECATED_processingGuarantees != null) processingGuarantees = DEPRECATED_processingGuarantees;
if (!StringUtils.isBlank(DEPRECATED_destinationTopicName)) destinationTopicName = DEPRECATED_destinationTopicName;
@@ -403,7 +406,6 @@ public class CmdSources extends CmdBase {
// if jar file is present locally then load jar and validate SinkClass in it
- ClassLoader classLoader = null;
if (archivePath != null) {
if (!fileExists(archivePath)) {
throw new ParameterException("Archive file " + archivePath + " does not exist");
@@ -436,14 +438,14 @@ public class CmdSources extends CmdBase {
throws IOException {
org.apache.pulsar.functions.proto.Function.FunctionDetails.Builder functionDetailsBuilder
= org.apache.pulsar.functions.proto.Function.FunctionDetails.newBuilder();
- Utils.mergeJson(FunctionsImpl.printJson(SourceConfigUtils.convert(sourceConfig)), functionDetailsBuilder);
+ Utils.mergeJson(FunctionsImpl.printJson(SourceConfigUtils.convert(sourceConfig, classLoader)), functionDetailsBuilder);
return functionDetailsBuilder.build();
}
protected String validateSourceType(String sourceType) throws IOException {
Set<String> availableSources;
try {
- availableSources = admin.functions().getSources();
+ availableSources = admin.source().getBuiltInSources().stream().map(ConnectorDefinition::getName).collect(Collectors.toSet());
} catch (PulsarAdminException e) {
throw new IOException(e);
}
@@ -487,16 +489,16 @@ public class CmdSources extends CmdBase {
@Override
void runCmd() throws Exception {
- admin.functions().deleteFunction(tenant, namespace, name);
+ admin.source().deleteSource(tenant, namespace, name);
print("Delete source successfully");
}
}
@Parameters(commandDescription = "Get the list of Pulsar IO connector sources supported by Pulsar cluster")
- public class ListSources extends BaseCommand {
+ public class ListBuiltInSources extends BaseCommand {
@Override
void runCmd() throws Exception {
- admin.functions().getConnectorsList().stream().filter(x -> !StringUtils.isEmpty(x.getSourceClass()))
+ admin.source().getBuiltInSources().stream().filter(x -> !StringUtils.isEmpty(x.getSourceClass()))
.forEach(connector -> {
System.out.println(connector.getName());
System.out.println(WordUtils.wrap(connector.getDescription(), 80));
[pulsar] 03/13: Merge branch 'master' into
serverside_validation_endpoints
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit eacd45380448b32547234ebfec05004959f630c1
Merge: 48f5233 79dd772
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Thu Oct 4 20:55:28 2018 -0700
Merge branch 'master' into serverside_validation_endpoints
conf/broker.conf | 3 +
conf/proxy.conf | 8 +
distribution/server/src/assemble/LICENSE.bin.txt | 2 +-
managed-ledger-shaded/pom.xml | 18 +-
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 10 +-
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 1 +
pom.xml | 6 +-
.../AuthenticationProviderAthenz.java | 12 +-
.../AuthenticationProviderAthenzTest.java | 15 ++
.../apache/pulsar/broker/ServiceConfiguration.java | 10 +
pulsar-broker-shaded/pom.xml | 19 +-
.../pulsar/broker/BookKeeperClientFactoryImpl.java | 2 +-
.../apache/pulsar/broker/SLAMonitoringTest.java | 2 +-
.../AntiAffinityNamespaceGroupTest.java | 2 +-
.../broker/loadbalance/LoadBalancerTest.java | 2 +-
.../loadbalance/ModularLoadManagerImplTest.java | 2 +-
.../loadbalance/SimpleLoadManagerImplTest.java | 2 +-
.../broker/service/AdvertisedAddressTest.java | 3 +-
.../broker/service/BacklogQuotaManagerTest.java | 9 +-
.../broker/service/BrokerBkEnsemblesTests.java | 10 +-
.../pulsar/broker/service/ReplicatorTestBase.java | 6 +-
.../broker/service/v1/V1_ReplicatorTestBase.java | 6 +-
.../pulsar/client/api/NonPersistentTopicTest.java | 6 +-
.../worker/PulsarWorkerAssignmentTest.java | 2 +-
.../apache/pulsar/io/PulsarFunctionAdminTest.java | 8 +-
.../apache/pulsar/io/PulsarFunctionTlsTest.java | 2 +-
.../org/apache/pulsar/io/PulsarSinkE2ETest.java | 2 +-
pulsar-client-admin-shaded/pom.xml | 19 +-
.../pulsar-client-kafka-shaded/pom.xml | 28 ---
.../client/kafka/compat/tests/KafkaApiTest.java | 113 -----------
.../pulsar-client-kafka/pom.xml | 29 +--
.../kafka/compat/tests/KafkaProducerTest.java | 119 -----------
pulsar-client-schema/pom.xml | 23 +--
pulsar-client-shaded/pom.xml | 23 +--
pulsar-common/pom.xml | 4 +-
.../pulsar/functions/api/examples/pojo/Tick.java | 40 +---
pulsar-functions/runtime-all/pom.xml | 10 +-
.../functions/runtime/KubernetesRuntimeTest.java | 7 +-
pulsar-functions/worker/pom.xml | 8 +-
pulsar-spark/pom.xml | 3 +
pulsar-sql/presto-distribution/LICENSE | 10 +
.../pulsar/sql/presto/TestPulsarConnector.java | 6 +-
.../pulsar/websocket/service/ProxyServer.java | 12 +-
pulsar-zookeeper-utils/pom.xml | 8 +
.../pulsar/zookeeper/LocalBookkeeperEnsemble.java | 41 +++-
.../zookeeper/LocalBookkeeperEnsembleTest.java | 16 +-
site2/website/data/team.js | 64 +++---
site2/website/pages/en/team.js | 23 ---
tests/integration/pom.xml | 9 +-
.../integration/compat/kafka/KafkaApiTest.java | 223 ++++++++++++++++-----
.../tests/integration/presto/TestBasicPresto.java | 61 +++---
.../integration/topologies/PulsarCluster.java | 40 +++-
.../src/test/resources/pulsar-process.xml | 1 +
53 files changed, 494 insertions(+), 616 deletions(-)
[pulsar] 01/13: First cut of adding new endpoints for source/sink
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ee81c7e7ad6a7b4f466ba4abf478545e6d3bccf2
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Thu Oct 4 16:56:26 2018 -0700
First cut of adding new endpoints for source/sink
---
.../pulsar/broker/admin/impl/FunctionsBase.java | 4 +-
.../apache/pulsar/broker/admin/impl/SinkBase.java | 258 ++++++++++++++++++++
.../pulsar/broker/admin/impl/SourceBase.java | 261 +++++++++++++++++++++
.../org/apache/pulsar/broker/admin/v1/Sink.java | 29 +++
.../org/apache/pulsar/broker/admin/v1/Source.java | 29 +++
.../org/apache/pulsar/broker/admin/v2/Sink.java | 34 +++
.../org/apache/pulsar/broker/admin/v2/Source.java | 34 +++
.../runtime/KubernetesRuntimeFactory.java | 3 +-
.../src/main/resources/java_instance_log4j2.yml | 4 +-
.../pulsar/functions/worker/rest/Resources.java | 4 +
.../functions/worker/rest/api/FunctionsImpl.java | 64 +++--
.../worker/rest/api/v2/FunctionApiV2Resource.java | 4 +-
.../worker/rest/api/v2/SinkApiV2Resource.java | 180 ++++++++++++++
.../worker/rest/api/v2/SourceApiV2Resource.java | 180 ++++++++++++++
14 files changed, 1060 insertions(+), 28 deletions(-)
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 cff3c34..d20bc62 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
@@ -84,7 +84,7 @@ public class FunctionsBase extends AdminResource implements Supplier<WorkerServi
final @FormDataParam("functionConfig") String functionConfigJson) {
return functions.registerFunction(tenant, namespace, functionName, uploadedInputStream, fileDetail,
- functionPkgUrl, functionDetailsJson, functionConfigJson, clientAppId());
+ functionPkgUrl, functionDetailsJson, functionConfigJson, null, null, clientAppId());
}
@PUT
@@ -106,7 +106,7 @@ public class FunctionsBase extends AdminResource implements Supplier<WorkerServi
final @FormDataParam("functionConfig") String functionConfigJson) {
return functions.updateFunction(tenant, namespace, functionName, uploadedInputStream, fileDetail,
- functionPkgUrl, functionDetailsJson, functionConfigJson, clientAppId());
+ functionPkgUrl, functionDetailsJson, functionConfigJson, null, null, clientAppId());
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinkBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinkBase.java
new file mode 100644
index 0000000..43ca8bf
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinkBase.java
@@ -0,0 +1,258 @@
+/**
+ * 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.broker.admin.impl;
+
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.broker.admin.AdminResource;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.functions.proto.Function.FunctionMetaData;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
+import org.apache.pulsar.functions.worker.WorkerService;
+import org.apache.pulsar.functions.worker.rest.api.FunctionsImpl;
+import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
+import org.glassfish.jersey.media.multipart.FormDataParam;
+
+import javax.ws.rs.*;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+
+public class SinkBase extends AdminResource implements Supplier<WorkerService> {
+
+ private final FunctionsImpl functions;
+
+ public SinkBase() {
+ this.functions = new FunctionsImpl(this);
+ }
+
+ @Override
+ public WorkerService get() {
+ return pulsar().getWorkerService();
+ }
+
+ @POST
+ @ApiOperation(value = "Creates a new Pulsar Sink in cluster mode")
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request (function already exists, etc.)"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 200, message = "Pulsar Function successfully created")
+ })
+ @Path("/{tenant}/{namespace}/{sinkName}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response registerSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String functionPkgUrl,
+ final @FormDataParam("sinkConfig") String sinkConfigJson) {
+
+ return functions.registerFunction(tenant, namespace, sinkName, uploadedInputStream, fileDetail,
+ functionPkgUrl, null, null, null, sinkConfigJson, clientAppId());
+ }
+
+ @PUT
+ @ApiOperation(value = "Updates a Pulsar Sink currently running in cluster mode")
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request (function doesn't exist, etc.)"),
+ @ApiResponse(code = 200, message = "Pulsar Function successfully updated")
+ })
+ @Path("/{tenant}/{namespace}/{sinkName}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response updateSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String functionPkgUrl,
+ final @FormDataParam("sinkConfig") String sinkConfigJson) {
+
+ return functions.updateFunction(tenant, namespace, sinkName, uploadedInputStream, fileDetail,
+ functionPkgUrl, null, null, null, sinkConfigJson, clientAppId());
+
+ }
+
+
+ @DELETE
+ @ApiOperation(value = "Deletes a Pulsar Sink currently running in cluster mode")
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function doesn't exist"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 200, message = "The function was successfully deleted")
+ })
+ @Path("/{tenant}/{namespace}/{sinkName}")
+ public Response deregisterSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName) {
+ return functions.deregisterFunction(tenant, namespace, sinkName, clientAppId());
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Fetches information about a Pulsar Sink currently running in cluster mode",
+ response = FunctionMetaData.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 404, message = "The function doesn't exist")
+ })
+ @Path("/{tenant}/{namespace}/{sinkName}")
+ public Response getSinkInfo(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName) throws IOException {
+ return functions.getFunctionInfo(tenant, namespace, sinkName);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Displays the status of a Pulsar Sink instance",
+ response = FunctionStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 404, message = "The function doesn't exist")
+ })
+ @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/status")
+ public Response getSinkInstanceStatus(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName,
+ final @PathParam("instanceId") String instanceId) throws IOException {
+ return functions.getFunctionInstanceStatus(
+ tenant, namespace, sinkName, instanceId, uri.getRequestUri());
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Displays the status of a Pulsar Sink running in cluster mode",
+ response = FunctionStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions")
+ })
+ @Path("/{tenant}/{namespace}/{sinkName}/status")
+ public Response getSinkStatus(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName) throws IOException {
+ return functions.getFunctionStatus(tenant, namespace, sinkName);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Lists all Pulsar Sinks currently deployed in a given namespace",
+ response = String.class,
+ responseContainer = "Collection"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions")
+ })
+ @Path("/{tenant}/{namespace}")
+ public Response listSinks(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace) {
+ return functions.listFunctions(tenant, namespace);
+
+ }
+
+ @POST
+ @ApiOperation(value = "Restart sink instance", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/restart")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response restartSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sinkName") String sinkName,
+ final @PathParam("instanceId") String instanceId) {
+ return functions.restartFunctionInstance(tenant, namespace, sinkName, instanceId, uri.getRequestUri());
+ }
+
+ @POST
+ @ApiOperation(value = "Restart all sink instances", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sinkName}/restart")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response restartSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sinkName") String sinkName) {
+ return functions.restartFunctionInstances(tenant, namespace, sinkName);
+ }
+
+ @POST
+ @ApiOperation(value = "Stop sink instance", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/stop")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response stopSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sinkName") String sinkName,
+ final @PathParam("instanceId") String instanceId) {
+ return functions.stopFunctionInstance(tenant, namespace, sinkName, instanceId, uri.getRequestUri());
+ }
+
+ @POST
+ @ApiOperation(value = "Stop all sink instances", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sinkName}/stop")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response stopSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sinkName") String sinkName) {
+ return functions.stopFunctionInstances(tenant, namespace, sinkName);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Fetches a list of supported Pulsar IO sink connectors currently running in cluster mode",
+ response = List.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 408, message = "Request timeout")
+ })
+ @Path("/builtinsinks")
+ public List<ConnectorDefinition> getSinkList() {
+ List<ConnectorDefinition> connectorDefinitions = functions.getListOfConnectors();
+ List<ConnectorDefinition> retval = new ArrayList<>();
+ for (ConnectorDefinition connectorDefinition : connectorDefinitions) {
+ if (!StringUtils.isEmpty(connectorDefinition.getSinkClass())) {
+ retval.add(connectorDefinition);
+ }
+ }
+ return retval;
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourceBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourceBase.java
new file mode 100644
index 0000000..b4428ab
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourceBase.java
@@ -0,0 +1,261 @@
+/**
+ * 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.broker.admin.impl;
+
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.broker.admin.AdminResource;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.functions.proto.Function.FunctionMetaData;
+import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
+import org.apache.pulsar.functions.worker.WorkerService;
+import org.apache.pulsar.functions.worker.rest.api.FunctionsImpl;
+import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
+import org.glassfish.jersey.media.multipart.FormDataParam;
+
+import javax.ws.rs.*;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+
+public class SourceBase extends AdminResource implements Supplier<WorkerService> {
+
+ private final FunctionsImpl functions;
+
+ public SourceBase() {
+ this.functions = new FunctionsImpl(this);
+ }
+
+ @Override
+ public WorkerService get() {
+ return pulsar().getWorkerService();
+ }
+
+ @POST
+ @ApiOperation(value = "Creates a new Pulsar Source in cluster mode")
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request (function already exists, etc.)"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 200, message = "Pulsar Function successfully created")
+ })
+ @Path("/{tenant}/{namespace}/{sourceName}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response registerSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String functionPkgUrl,
+ final @FormDataParam("sourceConfig") String sourceConfigJson) {
+
+ return functions.registerFunction(tenant, namespace, sourceName, uploadedInputStream, fileDetail,
+ functionPkgUrl, null, null, sourceConfigJson, null, clientAppId());
+ }
+
+ @PUT
+ @ApiOperation(value = "Updates a Pulsar Source currently running in cluster mode")
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request (function doesn't exist, etc.)"),
+ @ApiResponse(code = 200, message = "Pulsar Function successfully updated")
+ })
+ @Path("/{tenant}/{namespace}/{sourceName}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response updateSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String functionPkgUrl,
+ final @FormDataParam("sourceConfig") String sourceConfigJson) {
+
+ return functions.updateFunction(tenant, namespace, sourceName, uploadedInputStream, fileDetail,
+ functionPkgUrl, null, null, sourceConfigJson, null, clientAppId());
+
+ }
+
+
+ @DELETE
+ @ApiOperation(value = "Deletes a Pulsar Source currently running in cluster mode")
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function doesn't exist"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 200, message = "The function was successfully deleted")
+ })
+ @Path("/{tenant}/{namespace}/{sourceName}")
+ public Response deregisterSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName) {
+ return functions.deregisterFunction(tenant, namespace, sourceName, clientAppId());
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Fetches information about a Pulsar Source currently running in cluster mode",
+ response = FunctionMetaData.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 404, message = "The function doesn't exist")
+ })
+ @Path("/{tenant}/{namespace}/{sourceName}")
+ public Response getSourceInfo(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName) throws IOException {
+ return functions.getFunctionInfo(
+ tenant, namespace, sourceName);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Displays the status of a Pulsar Source instance",
+ response = FunctionStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 404, message = "The function doesn't exist")
+ })
+ @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/status")
+ public Response getSourceInstanceStatus(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName,
+ final @PathParam("instanceId") String instanceId) throws IOException {
+ return functions.getFunctionInstanceStatus(
+ tenant, namespace, sourceName, instanceId, uri.getRequestUri());
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Displays the status of a Pulsar Source running in cluster mode",
+ response = FunctionStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions")
+ })
+ @Path("/{tenant}/{namespace}/{sourceName}/status")
+ public Response getSourceStatus(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName) throws IOException {
+ return functions.getFunctionStatus(tenant, namespace, sourceName);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Lists all Pulsar Sources currently deployed in a given namespace",
+ response = String.class,
+ responseContainer = "Collection"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions")
+ })
+ @Path("/{tenant}/{namespace}")
+ public Response listSources(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace) {
+ return functions.listFunctions(
+ tenant, namespace);
+
+ }
+
+ @POST
+ @ApiOperation(value = "Restart source instance", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/restart")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response restartSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sourceName") String sourceName,
+ final @PathParam("instanceId") String instanceId) {
+ return functions.restartFunctionInstance(tenant, namespace, sourceName, instanceId, uri.getRequestUri());
+ }
+
+ @POST
+ @ApiOperation(value = "Restart all source instances", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sourceName}/restart")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response restartSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sourceName") String sourceName) {
+ return functions.restartFunctionInstances(tenant, namespace, sourceName);
+ }
+
+ @POST
+ @ApiOperation(value = "Stop source instance", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/stop")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response stopSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sourceName") String sourceName,
+ final @PathParam("instanceId") String instanceId) {
+ return functions.stopFunctionInstance(tenant, namespace, sourceName, instanceId, uri.getRequestUri());
+ }
+
+ @POST
+ @ApiOperation(value = "Stop all source instances", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sourceName}/stop")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response stopSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sourceName") String sourceName) {
+ return functions.stopFunctionInstances(tenant, namespace, sourceName);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Fetches a list of supported Pulsar IO source connectors currently running in cluster mode",
+ response = List.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 408, message = "Request timeout")
+ })
+ @Path("/builtinsources")
+ public List<ConnectorDefinition> getSourceList() {
+ List<ConnectorDefinition> connectorDefinitions = functions.getListOfConnectors();
+ List<ConnectorDefinition> retval = new ArrayList<>();
+ for (ConnectorDefinition connectorDefinition : connectorDefinitions) {
+ if (!StringUtils.isEmpty(connectorDefinition.getSourceClass())) {
+ retval.add(connectorDefinition);
+ }
+ }
+ return retval;
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Sink.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Sink.java
new file mode 100644
index 0000000..3d73d4d
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Sink.java
@@ -0,0 +1,29 @@
+/**
+ * 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.broker.admin.v1;
+
+import io.swagger.annotations.Api;
+import org.apache.pulsar.broker.admin.impl.SinkBase;
+
+import javax.ws.rs.Path;
+
+@Path("/sink")
+@Api(value = "/sink", description = "Sink admin apis", tags = "sink", hidden = true)
+public class Sink extends SinkBase {
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Source.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Source.java
new file mode 100644
index 0000000..99b41dc
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Source.java
@@ -0,0 +1,29 @@
+/**
+ * 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.broker.admin.v1;
+
+import io.swagger.annotations.Api;
+import org.apache.pulsar.broker.admin.impl.SourceBase;
+
+import javax.ws.rs.Path;
+
+@Path("/source")
+@Api(value = "/source", description = "Source admin apis", tags = "source", hidden = true)
+public class Source extends SourceBase {
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Sink.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Sink.java
new file mode 100644
index 0000000..aea0ae7
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Sink.java
@@ -0,0 +1,34 @@
+/**
+ * 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.broker.admin.v2;
+
+import io.swagger.annotations.Api;
+import org.apache.pulsar.broker.admin.impl.SinkBase;
+
+import javax.ws.rs.Consumes;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+
+@Path("/sink")
+@Api(value = "/sink", description = "Sink admin apis", tags = "sink", hidden = true)
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+public class Sink extends SinkBase {
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Source.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Source.java
new file mode 100644
index 0000000..e5ef56c
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Source.java
@@ -0,0 +1,34 @@
+/**
+ * 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.broker.admin.v2;
+
+import io.swagger.annotations.Api;
+import org.apache.pulsar.broker.admin.impl.SourceBase;
+
+import javax.ws.rs.Consumes;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+
+@Path("/source")
+@Api(value = "/source", description = "Source admin apis", tags = "source", hidden = true)
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+public class Source extends SourceBase {
+}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
index 8ba7505..cba9ebf 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
@@ -141,7 +141,8 @@ public class KubernetesRuntimeFactory implements RuntimeFactory {
KubernetesRuntime.doChecks(functionDetails);
}
- private void setupClient() throws Exception {
+ @VisibleForTesting
+ void setupClient() throws Exception {
if (appsClient == null) {
if (k8Uri == null) {
log.info("k8Uri is null thus going by defaults");
diff --git a/pulsar-functions/runtime/src/main/resources/java_instance_log4j2.yml b/pulsar-functions/runtime/src/main/resources/java_instance_log4j2.yml
index df25367..9846f05 100644
--- a/pulsar-functions/runtime/src/main/resources/java_instance_log4j2.yml
+++ b/pulsar-functions/runtime/src/main/resources/java_instance_log4j2.yml
@@ -44,7 +44,7 @@ Configuration:
name: RollingFile
fileName: "${sys:pulsar.function.log.dir}/${sys:pulsar.function.log.file}.log"
filePattern: "${sys:pulsar.function.log.dir}/${sys:pulsar.function.log.file}-%d{MM-dd-yyyy}-%i.log.gz"
- immediateFlush: false
+ immediateFlush: true
PatternLayout:
Pattern: "%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"
Policies:
@@ -72,7 +72,7 @@ Configuration:
name: BkRollingFile
fileName: "${sys:pulsar.function.log.dir}/${sys:pulsar.function.log.file}.bk"
filePattern: "${sys:pulsar.function.log.dir}/${sys:pulsar.function.log.file}.bk-%d{MM-dd-yyyy}-%i.log.gz"
- immediateFlush: false
+ immediateFlush: true
PatternLayout:
Pattern: "%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"
Policies:
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/Resources.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/Resources.java
index 26c7127..ac011db 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/Resources.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/Resources.java
@@ -20,6 +20,8 @@ package org.apache.pulsar.functions.worker.rest;
import org.apache.pulsar.functions.worker.rest.api.FunctionsMetricsResource;
import org.apache.pulsar.functions.worker.rest.api.v2.FunctionApiV2Resource;
+import org.apache.pulsar.functions.worker.rest.api.v2.SinkApiV2Resource;
+import org.apache.pulsar.functions.worker.rest.api.v2.SourceApiV2Resource;
import org.apache.pulsar.functions.worker.rest.api.v2.WorkerApiV2Resource;
import org.glassfish.jersey.media.multipart.MultiPartFeature;
@@ -36,6 +38,8 @@ public final class Resources {
return new HashSet<>(
Arrays.asList(
FunctionApiV2Resource.class,
+ SourceApiV2Resource.class,
+ SinkApiV2Resource.class,
WorkerApiV2Resource.class,
MultiPartFeature.class
));
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 3124731..0d30e37 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
@@ -27,7 +27,6 @@ import static org.apache.pulsar.functions.utils.Reflections.loadJar;
import static org.apache.pulsar.functions.utils.Utils.FILE;
import static org.apache.pulsar.functions.utils.Utils.HTTP;
import static org.apache.pulsar.functions.utils.Utils.isFunctionPackageUrlSupported;
-import static org.apache.pulsar.functions.utils.functioncache.FunctionClassLoaders.create;
import com.google.gson.Gson;
@@ -39,10 +38,7 @@ import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
-import java.net.URLClassLoader;
-import java.nio.file.CopyOption;
import java.nio.file.Files;
-import java.nio.file.NoSuchFileException;
import java.util.Base64;
import java.util.Collection;
import java.util.LinkedList;
@@ -89,9 +85,7 @@ import org.apache.pulsar.functions.proto.Function.SinkSpec;
import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
-import org.apache.pulsar.functions.utils.FunctionConfig;
-import org.apache.pulsar.functions.utils.FunctionConfigUtils;
-import org.apache.pulsar.functions.utils.functioncache.FunctionClassLoaders;
+import org.apache.pulsar.functions.utils.*;
import org.apache.pulsar.functions.utils.validation.ConfigValidation;
import org.apache.pulsar.functions.worker.FunctionMetaDataManager;
import org.apache.pulsar.functions.worker.FunctionRuntimeManager;
@@ -136,6 +130,7 @@ public class FunctionsImpl {
public Response registerFunction(final String tenant, final String namespace, final String functionName,
final InputStream uploadedInputStream, final FormDataContentDisposition fileDetail,
final String functionPkgUrl, final String functionDetailsJson, final String functionConfigJson,
+ final String sourceConfigJson, final String sinkConfigJson,
final String clientRole) {
if (!isWorkerServiceAvailable()) {
@@ -173,10 +168,10 @@ public class FunctionsImpl {
try {
if (isPkgUrlProvided) {
functionDetails = validateUpdateRequestParamsWithPkgUrl(tenant, namespace, functionName, functionPkgUrl,
- functionDetailsJson, functionConfigJson);
+ functionDetailsJson, functionConfigJson, sourceConfigJson, sinkConfigJson);
} else {
functionDetails = validateUpdateRequestParams(tenant, namespace, functionName, uploadedInputStreamAsFile,
- fileDetail, functionDetailsJson, functionConfigJson);
+ fileDetail, functionDetailsJson, functionConfigJson, sourceConfigJson, sinkConfigJson);
}
} catch (Exception e) {
log.error("Invalid register function request @ /{}/{}/{}", tenant, namespace, functionName, e);
@@ -216,7 +211,7 @@ public class FunctionsImpl {
public Response updateFunction(final String tenant, final String namespace, final String functionName,
final InputStream uploadedInputStream, final FormDataContentDisposition fileDetail,
final String functionPkgUrl, final String functionDetailsJson, final String functionConfigJson,
- final String clientRole) {
+ final String sourceConfigJson, final String sinkConfigJson, final String clientRole) {
if (!isWorkerServiceAvailable()) {
return getUnavailableResponse();
@@ -252,10 +247,10 @@ public class FunctionsImpl {
try {
if (isPkgUrlProvided) {
functionDetails = validateUpdateRequestParamsWithPkgUrl(tenant, namespace, functionName, functionPkgUrl,
- functionDetailsJson, functionConfigJson);
+ functionDetailsJson, functionConfigJson, sourceConfigJson, sinkConfigJson);
} else {
functionDetails = validateUpdateRequestParams(tenant, namespace, functionName, uploadedInputStreamAsFile,
- fileDetail, functionDetailsJson, functionConfigJson);
+ fileDetail, functionDetailsJson, functionConfigJson, sourceConfigJson, sinkConfigJson);
}
} catch (Exception e) {
log.error("Invalid register function request @ /{}/{}/{}", tenant, namespace, functionName, e);
@@ -877,23 +872,24 @@ public class FunctionsImpl {
}
private FunctionDetails validateUpdateRequestParamsWithPkgUrl(String tenant, String namespace, String functionName,
- String functionPkgUrl, String functionDetailsJson, String functionConfigJson)
+ String functionPkgUrl, String functionDetailsJson, String functionConfigJson,
+ String sourceConfigJson, String sinkConfigJson)
throws IllegalArgumentException, IOException, URISyntaxException {
if (!isFunctionPackageUrlSupported(functionPkgUrl)) {
throw new IllegalArgumentException("Function Package url is not valid. supported url (http/https/file)");
}
FunctionDetails functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
- functionDetailsJson, functionConfigJson, functionPkgUrl, null);
+ functionDetailsJson, functionConfigJson, sourceConfigJson, sinkConfigJson, functionPkgUrl, null);
return functionDetails;
}
private FunctionDetails validateUpdateRequestParams(String tenant, String namespace, String functionName,
File uploadedInputStreamAsFile, FormDataContentDisposition fileDetail, String functionDetailsJson,
- String functionConfigJson)
+ String functionConfigJson, String sourceConfigJson, String sinkConfigJson)
throws IllegalArgumentException, IOException, URISyntaxException {
FunctionDetails functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
- functionDetailsJson, functionConfigJson, null, uploadedInputStreamAsFile);
+ functionDetailsJson, functionConfigJson, sourceConfigJson, sinkConfigJson, null, uploadedInputStreamAsFile);
if (!isFunctionCodeBuiltin(functionDetails) && (uploadedInputStreamAsFile == null || fileDetail == null)) {
throw new IllegalArgumentException("Function Package is not provided");
}
@@ -966,7 +962,8 @@ public class FunctionsImpl {
}
private FunctionDetails validateUpdateRequestParams(String tenant, String namespace, String functionName,
- String functionDetailsJson, String functionConfigJson, String functionPkgUrl, File uploadedInputStreamAsFile) throws IOException, URISyntaxException {
+ String functionDetailsJson, String functionConfigJson, String sourceConfigJson,
+ String sinkConfigJson, String functionPkgUrl, File uploadedInputStreamAsFile) throws IOException, URISyntaxException {
if (tenant == null) {
throw new IllegalArgumentException("Tenant is not provided");
}
@@ -977,11 +974,24 @@ public class FunctionsImpl {
throw new IllegalArgumentException("Function Name is not provided");
}
- if (StringUtils.isEmpty(functionDetailsJson) && StringUtils.isEmpty(functionConfigJson)) {
- throw new IllegalArgumentException("FunctionConfig is not provided");
+ int numDefinitions = 0;
+ if (StringUtils.isEmpty(functionDetailsJson)) {
+ numDefinitions++;
+ }
+ if (StringUtils.isEmpty(functionConfigJson)) {
+ numDefinitions++;
+ }
+ if (StringUtils.isEmpty(sourceConfigJson)) {
+ numDefinitions++;
}
- if (!StringUtils.isEmpty(functionDetailsJson) && !StringUtils.isEmpty(functionConfigJson)) {
- throw new IllegalArgumentException("Only one of FunctionDetails or FunctionConfig should be provided");
+ if (StringUtils.isEmpty(sinkConfigJson)) {
+ numDefinitions++;
+ }
+ if (numDefinitions == 0) {
+ throw new IllegalArgumentException("Function Info is not provided");
+ }
+ if (numDefinitions > 1) {
+ throw new IllegalArgumentException("Conflicting Info provided");
}
if (!StringUtils.isEmpty(functionConfigJson)) {
FunctionConfig functionConfig = new Gson().fromJson(functionConfigJson, FunctionConfig.class);
@@ -995,6 +1005,18 @@ public class FunctionsImpl {
ConfigValidation.validateConfig(functionConfig, functionConfig.getRuntime().name(), clsLoader);
return FunctionConfigUtils.convert(functionConfig, clsLoader);
}
+ if (!StringUtils.isEmpty(sourceConfigJson)) {
+ SourceConfig sourceConfig = new Gson().fromJson(sourceConfigJson, SourceConfig.class);
+ ClassLoader clsLoader = extractClassLoader(functionPkgUrl, uploadedInputStreamAsFile);
+ ConfigValidation.validateConfig(sourceConfig, FunctionConfig.Runtime.JAVA.name(), clsLoader);
+ return SourceConfigUtils.convert(sourceConfig, clsLoader);
+ }
+ if (!StringUtils.isEmpty(sinkConfigJson)) {
+ SinkConfig sinkConfig = new Gson().fromJson(sinkConfigJson, SinkConfig.class);
+ ClassLoader clsLoader = extractClassLoader(functionPkgUrl, uploadedInputStreamAsFile);
+ ConfigValidation.validateConfig(sinkConfig, FunctionConfig.Runtime.JAVA.name(), clsLoader);
+ return SinkConfigUtils.convert(sinkConfig, clsLoader);
+ }
FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
org.apache.pulsar.functions.utils.Utils.mergeJson(functionDetailsJson, functionDetailsBuilder);
if (isNotBlank(functionPkgUrl)) {
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 1e44a60..a842b9e 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
@@ -60,7 +60,7 @@ public class FunctionApiV2Resource extends FunctionApiResource {
final @FormDataParam("functionConfig") String functionConfigJson) {
return functions.registerFunction(tenant, namespace, functionName, uploadedInputStream, fileDetail,
- functionPkgUrl, functionDetailsJson, functionConfigJson, clientAppId());
+ functionPkgUrl, functionDetailsJson, functionConfigJson, null, null, clientAppId());
}
@@ -77,7 +77,7 @@ public class FunctionApiV2Resource extends FunctionApiResource {
final @FormDataParam("functionConfig") String functionConfigJson) {
return functions.updateFunction(tenant, namespace, functionName, uploadedInputStream, fileDetail,
- functionPkgUrl, functionDetailsJson, functionConfigJson, clientAppId());
+ functionPkgUrl, functionDetailsJson, functionConfigJson, null, null, clientAppId());
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2Resource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2Resource.java
new file mode 100644
index 0000000..175e2eb
--- /dev/null
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2Resource.java
@@ -0,0 +1,180 @@
+/**
+ * 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.rest.api.v2;
+
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.functions.worker.rest.FunctionApiResource;
+import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
+import org.glassfish.jersey.media.multipart.FormDataParam;
+
+import javax.ws.rs.*;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@Slf4j
+@Path("/source")
+public class SinkApiV2Resource extends FunctionApiResource {
+
+ @POST
+ @Path("/{tenant}/{namespace}/{sinkName}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response registerSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String sourcePkgUrl,
+ final @FormDataParam("sinkConfig") String sinkConfigJson) {
+
+ return functions.registerFunction(tenant, namespace, sinkName, uploadedInputStream, fileDetail,
+ sourcePkgUrl, null, null, null, sinkConfigJson, clientAppId());
+
+ }
+
+ @PUT
+ @Path("/{tenant}/{namespace}/{sinkName}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response updateSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String functionPkgUrl,
+ final @FormDataParam("sinkConfig") String sinkConfigJson) {
+
+ return functions.updateFunction(tenant, namespace, sinkName, uploadedInputStream, fileDetail,
+ functionPkgUrl, null, null, null, sinkConfigJson, clientAppId());
+
+ }
+
+
+ @DELETE
+ @Path("/{tenant}/{namespace}/{sinkName}")
+ public Response deregisterSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sinkName") String sinkName) {
+ return functions.deregisterFunction(tenant, namespace, sinkName, clientAppId());
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}/{sinkName}")
+ public Response getSinkInfo(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName)
+ throws IOException {
+ return functions.getFunctionInfo(tenant, namespace, sinkName);
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/status")
+ public Response getSinkInstanceStatus(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName,
+ final @PathParam("instanceId") String instanceId) throws IOException {
+ return functions.getFunctionInstanceStatus(
+ tenant, namespace, sinkName, instanceId, uri.getRequestUri());
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}/{sinkName}/status")
+ public Response getSinkStatus(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sinkName") String sinkName) throws IOException {
+ return functions.getFunctionStatus(tenant, namespace, sinkName);
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}")
+ public Response listSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace) {
+ return functions.listFunctions(tenant, namespace);
+
+ }
+
+ @POST
+ @ApiOperation(value = "Restart sink instance", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/restart")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response restartSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sinkName") String sinkName,
+ final @PathParam("instanceId") String instanceId) {
+ return functions.restartFunctionInstance(tenant, namespace, sinkName, instanceId, this.uri.getRequestUri());
+ }
+
+ @POST
+ @ApiOperation(value = "Restart all sink instances", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sinkName}/restart")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response restartSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sinkName") String sinkName) {
+ return functions.restartFunctionInstances(tenant, namespace, sinkName);
+ }
+
+ @POST
+ @ApiOperation(value = "Stop sink instance", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/stop")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response stopSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sinkName") String sinkName,
+ final @PathParam("instanceId") String instanceId) {
+ return functions.stopFunctionInstance(tenant, namespace, sinkName, instanceId, this.uri.getRequestUri());
+ }
+
+ @POST
+ @ApiOperation(value = "Stop all sink instances", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sinkName}/stop")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response stopSink(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sinkName") String sinkName) {
+ return functions.stopFunctionInstances(tenant, namespace, sinkName);
+ }
+
+ @GET
+ @Path("/builtinsinks")
+ public List<ConnectorDefinition> getSinkList() {
+ List<ConnectorDefinition> connectorDefinitions = functions.getListOfConnectors();
+ List<ConnectorDefinition> retval = new ArrayList<>();
+ for (ConnectorDefinition connectorDefinition : connectorDefinitions) {
+ if (!StringUtils.isEmpty(connectorDefinition.getSinkClass())) {
+ retval.add(connectorDefinition);
+ }
+ }
+ return retval;
+ }
+}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2Resource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2Resource.java
new file mode 100644
index 0000000..9d25bf9
--- /dev/null
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2Resource.java
@@ -0,0 +1,180 @@
+/**
+ * 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.rest.api.v2;
+
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.functions.worker.rest.FunctionApiResource;
+import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
+import org.glassfish.jersey.media.multipart.FormDataParam;
+
+import javax.ws.rs.*;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@Slf4j
+@Path("/source")
+public class SourceApiV2Resource extends FunctionApiResource {
+
+ @POST
+ @Path("/{tenant}/{namespace}/{sourceName}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response registerSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String sourcePkgUrl,
+ final @FormDataParam("sourceConfig") String sourceConfigJson) {
+
+ return functions.registerFunction(tenant, namespace, sourceName, uploadedInputStream, fileDetail,
+ sourcePkgUrl, null, null, sourceConfigJson, null, clientAppId());
+
+ }
+
+ @PUT
+ @Path("/{tenant}/{namespace}/{sourceName}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response updateSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail,
+ final @FormDataParam("url") String functionPkgUrl,
+ final @FormDataParam("sourceConfig") String sourceConfigJson) {
+
+ return functions.updateFunction(tenant, namespace, sourceName, uploadedInputStream, fileDetail,
+ functionPkgUrl, null, null, sourceConfigJson, null, clientAppId());
+
+ }
+
+
+ @DELETE
+ @Path("/{tenant}/{namespace}/{sourceName}")
+ public Response deregisterSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sourceName") String sourceName) {
+ return functions.deregisterFunction(tenant, namespace, sourceName, clientAppId());
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}/{sourceName}")
+ public Response getSourceInfo(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName)
+ throws IOException {
+ return functions.getFunctionInfo(tenant, namespace, sourceName);
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/status")
+ public Response getSourceInstanceStatus(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName,
+ final @PathParam("instanceId") String instanceId) throws IOException {
+ return functions.getFunctionInstanceStatus(
+ tenant, namespace, sourceName, instanceId, uri.getRequestUri());
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}/{sourceName}/status")
+ public Response getSourceStatus(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("sourceName") String sourceName) throws IOException {
+ return functions.getFunctionStatus(tenant, namespace, sourceName);
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}")
+ public Response listSources(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace) {
+ return functions.listFunctions(tenant, namespace);
+
+ }
+
+ @POST
+ @ApiOperation(value = "Restart source instance", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/restart")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response restartSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sourceName") String sourceName,
+ final @PathParam("instanceId") String instanceId) {
+ return functions.restartFunctionInstance(tenant, namespace, sourceName, instanceId, this.uri.getRequestUri());
+ }
+
+ @POST
+ @ApiOperation(value = "Restart all source instances", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sourceName}/restart")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response restartSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sourceName") String sourceName) {
+ return functions.restartFunctionInstances(tenant, namespace, sourceName);
+ }
+
+ @POST
+ @ApiOperation(value = "Stop source instance", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/stop")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response stopSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sourceName") String sourceName,
+ final @PathParam("instanceId") String instanceId) {
+ return functions.stopFunctionInstance(tenant, namespace, sourceName, instanceId, this.uri.getRequestUri());
+ }
+
+ @POST
+ @ApiOperation(value = "Stop all source instances", response = Void.class)
+ @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error") })
+ @Path("/{tenant}/{namespace}/{sourceName}/stop")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response stopSource(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace, final @PathParam("sourceName") String sourceName) {
+ return functions.stopFunctionInstances(tenant, namespace, sourceName);
+ }
+
+ @GET
+ @Path("/builtinsources")
+ public List<ConnectorDefinition> getSourceList() {
+ List<ConnectorDefinition> connectorDefinitions = functions.getListOfConnectors();
+ List<ConnectorDefinition> retval = new ArrayList<>();
+ for (ConnectorDefinition connectorDefinition : connectorDefinitions) {
+ if (!StringUtils.isEmpty(connectorDefinition.getSourceClass())) {
+ retval.add(connectorDefinition);
+ }
+ }
+ return retval;
+ }
+}
[pulsar] 07/13: Merge branch 'master' into
serverside_validation_endpoints
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch srkukarni/serverside_validation_endpoints
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e4f848ce171de4b0769bfdd56a146a08d524d194
Merge: 41b108a f8e7dfe
Author: Sanjeev Kulkarni <sa...@streaml.io>
AuthorDate: Sat Oct 6 10:29:26 2018 -0700
Merge branch 'master' into serverside_validation_endpoints
conf/log4j2.yaml | 14 ++++--
distribution/server/pom.xml | 5 ++
distribution/server/src/assemble/LICENSE.bin.txt | 1 +
pom.xml | 6 +++
.../apache/pulsar/functions/instance/Utils.java | 57 ++++++++++++++++++++++
.../pulsar/functions/source/PulsarRecord.java | 2 +-
pulsar-functions/instance/src/main/python/util.py | 5 +-
.../pulsar/functions/instance}/UtilsTest.java | 2 +-
.../functions/runtime/KubernetesRuntime.java | 2 +-
.../functions/runtime/KubernetesRuntimeTest.java | 2 +-
pulsar-functions/utils/pom.xml | 14 ++++--
.../pulsar/functions/utils/FunctionConfig.java | 2 -
.../functions/utils/FunctionConfigUtils.java | 1 -
.../apache/pulsar/functions/utils/Resources.java | 3 --
.../org/apache/pulsar/functions/utils/Utils.java | 25 ----------
.../validation/ConfigValidationAnnotations.java | 2 -
.../pulsar/functions/worker/SchedulerManager.java | 53 +++++++++++++++-----
.../functions/worker/SchedulerManagerTest.java | 15 +-----
.../apache/pulsar/io/jdbc/JdbcAbstractSink.java | 7 ++-
19 files changed, 140 insertions(+), 78 deletions(-)