You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by rd...@apache.org on 2018/07/13 22:24:40 UTC
[incubator-pulsar] branch master updated: Fix: file-url sink/source
submission with cli (#2150)
This is an automated email from the ASF dual-hosted git repository.
rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 3b8faea Fix: file-url sink/source submission with cli (#2150)
3b8faea is described below
commit 3b8faea91e3dcd264545b0a36c58ea28d4a8bbba
Author: Rajan Dhabalia <rd...@apache.org>
AuthorDate: Fri Jul 13 15:24:37 2018 -0700
Fix: file-url sink/source submission with cli (#2150)
---
.../java/org/apache/pulsar/admin/cli/CmdSinks.java | 23 +++++++++++++++------
.../org/apache/pulsar/admin/cli/CmdSources.java | 21 ++++++++++++++-----
.../src/test/resources/pulsar-io-cassandra.nar | Bin 0 -> 9588943 bytes
.../src/test/resources/pulsar-io-twitter.nar | Bin 0 -> 6537014 bytes
.../apache/pulsar/functions/utils/SinkConfig.java | 1 +
.../pulsar/functions/utils/SourceConfig.java | 1 +
6 files changed, 35 insertions(+), 11 deletions(-)
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 c27b40a..31f3e0a 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
@@ -224,6 +224,8 @@ public class CmdSinks extends CmdBase {
protected Integer parallelism;
@Parameter(names = {"-a", "--archive"}, description = "Path to the archive file for the sink. It also supports url-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package.", listConverter = StringConverter.class)
protected String archive;
+ @Parameter(names = "--className", description = "The sink's class name if archive is file-url-path (file://)")
+ protected String className;
@Parameter(names = "--sinkConfigFile", description = "The path to a YAML config file specifying the "
+ "sink's configuration")
@@ -257,6 +259,10 @@ public class CmdSinks extends CmdBase {
sinkConfig.setNamespace(namespace);
}
+ if (null != className) {
+ sinkConfig.setClassName(className);
+ }
+
if (null != name) {
sinkConfig.setName(name);
}
@@ -427,12 +433,17 @@ public class CmdSinks extends CmdBase {
boolean isBuiltin = sinkConfig.getArchive().startsWith(Utils.BUILTIN);
if (!isBuiltin) {
- sinkClassName = ConnectorUtils.getIOSinkClass(sinkConfig.getArchive());
-
- try (NarClassLoader ncl = NarClassLoader.getFromArchive(new File(sinkConfig.getArchive()),
- Collections.emptySet())) {
- typeArg = sinkConfig.getArchive().startsWith(Utils.FILE) ? null
- : Utils.getSinkType(sinkClassName, ncl).getName();
+ if (sinkConfig.getArchive().startsWith(Utils.FILE)) {
+ if (StringUtils.isBlank(sinkConfig.getClassName())) {
+ throw new ParameterException("Class-name must be present for archive with file-url");
+ }
+ 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();
+ }
}
}
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 7c147e5..d050ab5 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
@@ -216,6 +216,8 @@ public class CmdSources extends CmdBase {
@Parameter(names = { "-a", "--archive" },
description = "The path to the NAR archive for the Source. It also supports url-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package.", listConverter = StringConverter.class)
protected String archive;
+ @Parameter(names = "--className", description = "The source's class name if archive is file-url-path (file://)")
+ protected String className;
@Parameter(names = "--sourceConfigFile", description = "The path to a YAML config file specifying the "
+ "source's configuration")
protected String sourceConfigFile;
@@ -248,6 +250,9 @@ public class CmdSources extends CmdBase {
if (null != name) {
sourceConfig.setName(name);
}
+ if (null != className) {
+ this.sourceConfig.setClassName(className);
+ }
if (null != destinationTopicName) {
sourceConfig.setTopicName(destinationTopicName);
}
@@ -391,12 +396,18 @@ public class CmdSources extends CmdBase {
boolean isBuiltin = sourceConfig.getArchive().startsWith(Utils.BUILTIN);
if (!isBuiltin) {
- sourceClassName = ConnectorUtils.getIOSourceClass(sourceConfig.getArchive());
+ if (sourceConfig.getArchive().startsWith(Utils.FILE)) {
+ if (StringUtils.isBlank(sourceConfig.getClassName())) {
+ throw new ParameterException("Class-name must be present for archive with file-url");
+ }
+ 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 = sourceConfig.getArchive().startsWith(Utils.FILE) ? null
- : getSourceType(sourceClassName, ncl).getName();
+ try (NarClassLoader ncl = NarClassLoader.getFromArchive(new File(sourceConfig.getArchive()),
+ Collections.emptySet())) {
+ typeArg = getSourceType(sourceClassName, ncl).getName();
+ }
}
}
diff --git a/pulsar-client-tools/src/test/resources/pulsar-io-cassandra.nar b/pulsar-client-tools/src/test/resources/pulsar-io-cassandra.nar
new file mode 100644
index 0000000..c061f98
Binary files /dev/null and b/pulsar-client-tools/src/test/resources/pulsar-io-cassandra.nar differ
diff --git a/pulsar-client-tools/src/test/resources/pulsar-io-twitter.nar b/pulsar-client-tools/src/test/resources/pulsar-io-twitter.nar
new file mode 100644
index 0000000..7684c77
Binary files /dev/null and b/pulsar-client-tools/src/test/resources/pulsar-io-twitter.nar differ
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java
index 2984c6b..d97aab2 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java
@@ -48,6 +48,7 @@ public class SinkConfig {
private String namespace;
@NotNull
private String name;
+ private String className;
@isMapEntryCustom(keyValidatorClasses = { ValidatorImpls.TopicNameValidator.class },
valueValidatorClasses = { ValidatorImpls.SerdeValidator.class })
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 2ffbcc9..5c53a4f 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
@@ -48,6 +48,7 @@ public class SourceConfig {
private String namespace;
@NotNull
private String name;
+ private String className;
@NotNull
@isValidTopicName