You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/07/11 05:22:15 UTC
[GitHub] sijie closed pull request #2124: patching CmdSources and adding
unit tests
sijie closed pull request #2124: patching CmdSources and adding unit tests
URL: https://github.com/apache/incubator-pulsar/pull/2124
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-tools/pom.xml b/pulsar-client-tools/pom.xml
index 795b2041e3..f50662179f 100644
--- a/pulsar-client-tools/pom.xml
+++ b/pulsar-client-tools/pom.xml
@@ -94,6 +94,13 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-io-twitter</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
<build>
@@ -114,6 +121,9 @@
<echo>copy test sink package</echo>
<mkdir dir="${basedir}/src/test/resources"/>
<copy file="${basedir}/../pulsar-io/cassandra/target/pulsar-io-cassandra-${project.version}.nar" tofile="${basedir}/src/test/resources/pulsar-io-cassandra.nar"/>
+ <echo>copy test source package</echo>
+ <mkdir dir="${basedir}/src/test/resources"/>
+ <copy file="${basedir}/../pulsar-io/twitter/target/pulsar-io-twitter-${project.version}.nar" tofile="${basedir}/src/test/resources/pulsar-io-twitter.nar"/>
</tasks>
</configuration>
</execution>
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 2ddd09f46c..2c91588f05 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
@@ -304,7 +304,7 @@ protected void inferMissingArguments(SinkConfig sinkConfig) {
protected void validateSinkConfigs(SinkConfig sinkConfig) {
if (isBlank(sinkConfig.getArchive())) {
- throw new ParameterException("Sink jar not specfied");
+ throw new ParameterException("Sink archive not specfied");
}
boolean isArchivePathUrl = Utils.isFunctionPackageUrlSupported(sinkConfig.getArchive());
@@ -342,7 +342,7 @@ protected void validateSinkConfigs(SinkConfig sinkConfig) {
// Validate sink class
ConnectorUtils.getIOSinkClass(archivePath);
} catch (IOException e) {
- throw new ParameterException("Failed to validate connector from " + archivePath, e);
+ throw new ParameterException("Connector from " + archivePath + " has error: " + e.getMessage());
}
}
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 bd5134f3bd..53609d0c0b 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
@@ -18,30 +18,14 @@
*/
package org.apache.pulsar.admin.cli;
-import static org.apache.pulsar.common.naming.TopicName.DEFAULT_NAMESPACE;
-import static org.apache.pulsar.common.naming.TopicName.PUBLIC_TENANT;
-import static org.apache.pulsar.functions.utils.Utils.convertProcessingGuarantee;
-import static org.apache.pulsar.functions.utils.Utils.fileExists;
-import static org.apache.pulsar.functions.utils.Utils.getSourceType;
-import static org.apache.pulsar.functions.worker.Utils.downloadFromHttpUrl;
-
import com.beust.jcommander.Parameter;
import com.beust.jcommander.ParameterException;
import com.beust.jcommander.Parameters;
import com.beust.jcommander.converters.StringConverter;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
-
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.lang.reflect.Type;
-import java.util.Collections;
-import java.util.Map;
-
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
-
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.text.WordUtils;
import org.apache.pulsar.admin.cli.utils.CmdUtils;
@@ -61,6 +45,20 @@
import org.apache.pulsar.functions.utils.io.ConnectorUtils;
import org.apache.pulsar.functions.utils.validation.ConfigValidation;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.util.Collections;
+import java.util.Map;
+
+import static org.apache.pulsar.common.naming.TopicName.DEFAULT_NAMESPACE;
+import static org.apache.pulsar.common.naming.TopicName.PUBLIC_TENANT;
+import static org.apache.pulsar.functions.utils.Utils.convertProcessingGuarantee;
+import static org.apache.pulsar.functions.utils.Utils.fileExists;
+import static org.apache.pulsar.functions.utils.Utils.getSourceType;
+import static org.apache.pulsar.functions.worker.Utils.downloadFromHttpUrl;
+
@Getter
@Parameters(commandDescription = "Interface for managing Pulsar IO Sources (ingress data into Pulsar)")
@Slf4j
@@ -103,7 +101,7 @@ void processArguments() throws Exception {
}
@Parameters(commandDescription = "Run a Pulsar IO source connector locally (rather than deploying it to the Pulsar cluster)")
- class LocalSourceRunner extends CreateSource {
+ protected class LocalSourceRunner extends CreateSource {
@Parameter(names = "--brokerServiceUrl", description = "The URL for the Pulsar broker")
protected String brokerServiceUrl;
@@ -140,7 +138,7 @@ void runCmd() throws Exception {
}
@Parameters(commandDescription = "Submit a Pulsar IO source connector to run in a Pulsar cluster")
- public class CreateSource extends SourceCommand {
+ protected class CreateSource extends SourceCommand {
@Override
void runCmd() throws Exception {
if (Utils.isFunctionPackageUrlSupported(this.sourceConfig.getArchive())) {
@@ -153,7 +151,7 @@ void runCmd() throws Exception {
}
@Parameters(commandDescription = "Update a Pulsar IO source connector")
- public class UpdateSource extends SourceCommand {
+ protected class UpdateSource extends SourceCommand {
@Override
void runCmd() throws Exception {
if (Utils.isFunctionPackageUrlSupported(sourceConfig.getArchive())) {
@@ -234,17 +232,36 @@ void processArguments() throws Exception {
sourceConfig.setArchive(archive);
}
- sourceConfig.setResources(new org.apache.pulsar.functions.utils.Resources(cpu, ram, disk));
+ org.apache.pulsar.functions.utils.Resources resources = sourceConfig.getResources();
+ if (resources == null) {
+ resources = new org.apache.pulsar.functions.utils.Resources();
+ }
+ if (cpu != null) {
+ resources.setCpu(cpu);
+ }
+
+ if (ram != null) {
+ resources.setRam(ram);
+ }
+
+ if (disk != null) {
+ resources.setDisk(disk);
+ }
+ sourceConfig.setResources(resources);
if (null != sourceConfigString) {
- Type type = new TypeToken<Map<String, String>>(){}.getType();
- Map<String, Object> sourceConfigMap = new Gson().fromJson(sourceConfigString, type);
- sourceConfig.setConfigs(sourceConfigMap);
+ sourceConfig.setConfigs(parseConfigs(sourceConfigString));
}
inferMissingArguments(sourceConfig);
}
+ protected Map<String, Object> parseConfigs(String str) {
+ Type type = new TypeToken<Map<String, String>>() {
+ }.getType();
+ return new Gson().fromJson(str, type);
+ }
+
private void inferMissingArguments(SourceConfig sourceConfig) {
if (sourceConfig.getTenant() == null) {
sourceConfig.setTenant(PUBLIC_TENANT);
@@ -296,7 +313,7 @@ protected void validateSourceConfigs(SourceConfig sourceConfig) {
// Validate source class
ConnectorUtils.getIOSourceClass(archivePath);
} catch (IOException e) {
- throw new ParameterException("Failed to validate connector from " + archivePath, e);
+ throw new ParameterException("Connector from " + archivePath + " has error: " + e.getMessage());
}
}
@@ -304,8 +321,7 @@ protected void validateSourceConfigs(SourceConfig sourceConfig) {
// Need to load jar and set context class loader before calling
ConfigValidation.validateConfig(sourceConfig, FunctionConfig.Runtime.JAVA.name());
} catch (Exception e) {
- e.printStackTrace();
- throw new ParameterException(e);
+ throw new ParameterException(e.getMessage());
}
}
@@ -396,7 +412,7 @@ protected FunctionDetails createSourceConfig(SourceConfig sourceConfig) throws I
}
@Parameters(commandDescription = "Stops a Pulsar IO source connector")
- class DeleteSource extends BaseCommand {
+ protected class DeleteSource extends BaseCommand {
@Parameter(names = "--tenant", description = "The tenant of a sink or source")
protected String tenant;
@@ -411,7 +427,7 @@ protected FunctionDetails createSourceConfig(SourceConfig sourceConfig) throws I
void processArguments() throws Exception {
super.processArguments();
if (null == name) {
- throw new RuntimeException(
+ throw new ParameterException(
"You must specify a name for the source");
}
if (tenant == null) {
diff --git a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java
index 4e230d3e5a..d07c85ea7c 100644
--- a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java
+++ b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java
@@ -43,8 +43,6 @@
import java.net.URL;
import java.nio.file.Files;
import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
import java.util.Map;
import static org.apache.pulsar.common.naming.TopicName.DEFAULT_NAMESPACE;
@@ -77,11 +75,13 @@ public IObjectFactory getObjectFactory() {
= FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE;
private static final Integer PARALLELISM = 1;
private static final String JAR_FILE_NAME = "pulsar-io-cassandra.nar";
+ private static final String WRONG_JAR_FILE_NAME = "pulsar-io-twitter.nar";
private String JAR_FILE_PATH;
+ private String WRONG_JAR_PATH;
private static final Double CPU = 100.0;
private static final Long RAM = 1024L * 1024L;
private static final Long DISK = 1024L * 1024L * 1024L;
- private static final String SINK_CONFIG_STRING = "{\"created_at\":\"Mon Jul 02 00:33:15 +0000 2018\"}";
+ private static final String SINK_CONFIG_STRING = "{\"created_at\":\"Mon Jul 02 00:33:15 0000 2018\"}";
private PulsarAdmin pulsarAdmin;
private Functions functions;
@@ -111,6 +111,7 @@ public void setup() throws Exception {
throw new RuntimeException("Failed to file required test archive: " + JAR_FILE_NAME);
}
JAR_FILE_PATH = file.getFile();
+ WRONG_JAR_PATH = Thread.currentThread().getContextClassLoader().getResource(WRONG_JAR_FILE_NAME).getFile();
Thread.currentThread().setContextClassLoader(Reflections.loadJar(new File(JAR_FILE_PATH)));
}
@@ -141,7 +142,6 @@ public void testCliCorrect() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -164,7 +164,6 @@ public void testMissingTenant() throws Exception {
null,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -187,7 +186,6 @@ public void testMissingNamespace() throws Exception {
TENANT,
null,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -210,7 +208,6 @@ public void testMissingName() throws Exception {
TENANT,
NAMESPACE,
null,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -235,7 +232,6 @@ public void testMissingInput() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
null,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -260,7 +256,6 @@ public void testMissingCustomSerdeInput() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
null,
@@ -283,7 +278,6 @@ public void testMissingTopicPattern() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
null,
CUSTOM_SERDE_INPUT_STRING,
@@ -307,7 +301,6 @@ public void testMissingAllInputTopics() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
null,
null,
null,
@@ -330,7 +323,6 @@ public void testMissingProcessingGuarantees() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -353,7 +345,6 @@ public void testMissingParallelism() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -376,7 +367,6 @@ public void testNegativeParallelism() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -399,7 +389,6 @@ public void testZeroParallelism() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -414,7 +403,7 @@ public void testZeroParallelism() throws Exception {
);
}
- @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Sink jar not specfied")
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Sink archive not specfied")
public void testMissingArchive() throws Exception {
SinkConfig sinkConfig = getSinkConfig();
sinkConfig.setArchive(null);
@@ -422,7 +411,6 @@ public void testMissingArchive() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -437,7 +425,30 @@ public void testMissingArchive() throws Exception {
);
}
- @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Archive file /tmp/foo.jar does not exist")
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Connector from .*.pulsar-io-twitter.nar has error: The 'twitter' connector does not provide a sink implementation")
+ public void testInvalidJarWithNoSource() throws Exception {
+ SinkConfig sinkConfig = getSinkConfig();
+ sinkConfig.setArchive(WRONG_JAR_PATH);
+ testCmdSinkCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ INPUTS,
+ TOPIC_PATTERN,
+ CUSTOM_SERDE_INPUT_STRING,
+ PROCESSING_GUARANTEES,
+ PARALLELISM,
+ WRONG_JAR_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sinkConfig
+ );
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Archive file /tmp/foo.jar" +
+ " does not exist")
public void testInvalidJar() throws Exception {
SinkConfig sinkConfig = getSinkConfig();
String fakeJar = "/tmp/foo.jar";
@@ -446,7 +457,6 @@ public void testInvalidJar() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -469,7 +479,6 @@ public void testMissingCpu() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -492,7 +501,6 @@ public void testMissingRam() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -515,7 +523,6 @@ public void testMissingDisk() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -538,7 +545,6 @@ public void testMissingConfig() throws Exception {
TENANT,
NAMESPACE,
NAME,
- CLASS_NAME,
INPUTS,
TOPIC_PATTERN,
CUSTOM_SERDE_INPUT_STRING,
@@ -557,7 +563,6 @@ public void testCmdSinkCliMissingArgs(
String tenant,
String namespace,
String name,
- String className,
String inputs,
String topicPattern,
String customSerdeInputString,
@@ -751,7 +756,7 @@ public void testCmdSinkConfigFileMissingResources() throws Exception {
testCmdSinkConfigFile(testSinkConfig, expectedSinkConfig);
}
- @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Sink jar not specfied")
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Sink archive not specfied")
public void testCmdSinkConfigFileMissingJar() throws Exception {
SinkConfig testSinkConfig = getSinkConfig();
testSinkConfig.setArchive(null);
@@ -771,6 +776,16 @@ public void testCmdSinkConfigFileInvalidJar() throws Exception {
testCmdSinkConfigFile(testSinkConfig, expectedSinkConfig);
}
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Connector from .*.pulsar-io-twitter.nar has error: The 'twitter' connector does not provide a sink implementation")
+ public void testCmdSinkConfigFileInvalidJarNoSink() throws Exception {
+ SinkConfig testSinkConfig = getSinkConfig();
+ testSinkConfig.setArchive(WRONG_JAR_PATH);
+
+ SinkConfig expectedSinkConfig = getSinkConfig();
+ expectedSinkConfig.setArchive(WRONG_JAR_PATH);
+ testCmdSinkConfigFile(testSinkConfig, expectedSinkConfig);
+ }
+
public void testCmdSinkConfigFile(SinkConfig testSinkConfig, SinkConfig expectedSinkConfig) throws Exception {
File file = Files.createTempFile("", "").toFile();
diff --git a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java
new file mode 100644
index 0000000000..cd9605146a
--- /dev/null
+++ b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java
@@ -0,0 +1,858 @@
+/**
+ * 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.admin.cli;
+
+import com.beust.jcommander.ParameterException;
+import com.fasterxml.jackson.dataformat.yaml.YAMLMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.admin.cli.utils.CmdUtils;
+import org.apache.pulsar.client.admin.Functions;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.functions.api.utils.DefaultSerDe;
+import org.apache.pulsar.functions.utils.FunctionConfig;
+import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.functions.utils.Resources;
+import org.apache.pulsar.functions.utils.SourceConfig;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.testng.Assert;
+import org.testng.IObjectFactory;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.pulsar.common.naming.TopicName.DEFAULT_NAMESPACE;
+import static org.apache.pulsar.common.naming.TopicName.PUBLIC_TENANT;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.powermock.api.mockito.PowerMockito.mockStatic;
+
+@Slf4j
+@PrepareForTest({CmdFunctions.class})
+@PowerMockIgnore({ "javax.management.*", "javax.ws.*", "org.apache.logging.log4j.*", "org.apache.pulsar.io.core.*" })
+public class TestCmdSources {
+
+ @ObjectFactory
+ public IObjectFactory getObjectFactory() {
+ return new org.powermock.modules.testng.PowerMockObjectFactory();
+ }
+
+ private static final String TENANT = "test-tenant";
+ private static final String NAMESPACE = "test-namespace";
+ private static final String NAME = "test";
+ private static final String TOPIC_NAME = "src_topic_1";
+ private static final String SERDE_CLASS_NAME = DefaultSerDe.class.getName();
+ private static final FunctionConfig.ProcessingGuarantees PROCESSING_GUARANTEES
+ = FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE;
+ private static final Integer PARALLELISM = 1;
+ private static final String JAR_FILE_NAME = "pulsar-io-twitter.nar";
+ private static final String WRONG_JAR_FILE_NAME = "pulsar-io-cassandra.nar";
+ private String JAR_FILE_PATH;
+ private String WRONG_JAR_PATH;
+ private static final Double CPU = 100.0;
+ private static final Long RAM = 1024L * 1024L;
+ private static final Long DISK = 1024L * 1024L * 1024L;
+ private static final String SINK_CONFIG_STRING = "{\"created_at\":\"Mon Jul 02 00:33:15 +0000 2018\"}";
+
+ private PulsarAdmin pulsarAdmin;
+ private Functions functions;
+ private CmdSources CmdSources;
+ private CmdSources.CreateSource createSource;
+ private CmdSources.UpdateSource updateSource;
+ private CmdSources.LocalSourceRunner localSourceRunner;
+ private CmdSources.DeleteSource deleteSource;
+
+ @BeforeMethod
+ public void setup() throws Exception {
+
+ pulsarAdmin = mock(PulsarAdmin.class);
+ functions = mock(Functions.class);
+ when(pulsarAdmin.functions()).thenReturn(functions);
+
+ CmdSources = spy(new CmdSources(pulsarAdmin));
+ createSource = spy(CmdSources.getCreateSource());
+ updateSource = spy(CmdSources.getUpdateSource());
+ localSourceRunner = spy(CmdSources.getLocalSourceRunner());
+ deleteSource = spy(CmdSources.getDeleteSource());
+
+ mockStatic(CmdFunctions.class);
+ PowerMockito.doNothing().when(CmdFunctions.class, "startLocalRun", Mockito.any(), Mockito.anyInt(), Mockito.anyInt(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
+ JAR_FILE_PATH = Thread.currentThread().getContextClassLoader().getResource(JAR_FILE_NAME).getFile();
+ WRONG_JAR_PATH = Thread.currentThread().getContextClassLoader().getResource(WRONG_JAR_FILE_NAME).getFile();
+ Thread.currentThread().setContextClassLoader(Reflections.loadJar(new File(JAR_FILE_PATH)));
+ }
+
+ public SourceConfig getSourceConfig() {
+ SourceConfig sourceConfig = new SourceConfig();
+ sourceConfig.setTenant(TENANT);
+ sourceConfig.setNamespace(NAMESPACE);
+ sourceConfig.setName(NAME);
+
+ sourceConfig.setTopicName(TOPIC_NAME);
+ sourceConfig.setSerdeClassName(SERDE_CLASS_NAME);
+ sourceConfig.setProcessingGuarantees(PROCESSING_GUARANTEES);
+ sourceConfig.setParallelism(PARALLELISM);
+ sourceConfig.setArchive(JAR_FILE_PATH);
+ sourceConfig.setResources(new Resources(CPU, RAM, DISK));
+ sourceConfig.setConfigs(createSource.parseConfigs(SINK_CONFIG_STRING));
+ return sourceConfig;
+ }
+
+ @Test
+ public void testCliCorrect() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test
+ public void testMissingTenant() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setTenant(PUBLIC_TENANT);
+ testCmdSourceCliMissingArgs(
+ null,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test
+ public void testMissingNamespace() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setNamespace(DEFAULT_NAMESPACE);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ null,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Field 'name' cannot be null!")
+ public void testMissingName() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setName(null);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ null,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Field 'topicName' cannot be null!")
+ public void testMissingTopicName() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setTopicName(null);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ null,
+ SERDE_CLASS_NAME,
+ PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test
+ public void testMissingSerdeClassName() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setSerdeClassName(null);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME,
+ null,
+ PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test
+ public void testMissingProcessingGuarantees() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setProcessingGuarantees(null);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, null,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test
+ public void testMissingParallelism() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setParallelism(1);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ null,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Field 'parallelism' must be a Positive Number")
+ public void testNegativeParallelism() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setParallelism(-1);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ -1,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Field 'parallelism' must be a Positive Number")
+ public void testZeroParallelism() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setParallelism(0);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ 0,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Source archive not specfied")
+ public void testMissingArchive() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setArchive(null);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ null,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Archive file /tmp/foo.jar does not exist")
+ public void testInvalidJar() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ String fakeJar = "/tmp/foo.jar";
+ sourceConfig.setArchive(fakeJar);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ fakeJar,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Connector from .*.pulsar-io-cassandra.nar has error: The 'cassandra' connector does not provide a source implementation")
+ public void testInvalidJarWithNoSource() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setArchive(WRONG_JAR_PATH);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ WRONG_JAR_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test
+ public void testMissingCpu() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setResources(new Resources(null, RAM, DISK));
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ null,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test
+ public void testMissingRam() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setResources(new Resources(CPU, null, DISK));
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ null,
+ DISK,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test
+ public void testMissingDisk() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setResources(new Resources(CPU, RAM, null));
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ null,
+ SINK_CONFIG_STRING,
+ sourceConfig
+ );
+ }
+
+ @Test
+ public void testMissingConfig() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ sourceConfig.setConfigs(null);
+ testCmdSourceCliMissingArgs(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ null,
+ sourceConfig
+ );
+ }
+
+ public void testCmdSourceCliMissingArgs(
+ String tenant,
+ String namespace,
+ String name,
+ String topicName, String serdeClassName,
+ FunctionConfig.ProcessingGuarantees processingGuarantees,
+ Integer parallelism,
+ String jarFile,
+ Double cpu,
+ Long ram,
+ Long disk,
+ String sourceConfigString,
+ SourceConfig sourceConfig) throws Exception {
+
+ // test create source
+ createSource.tenant = tenant;
+ createSource.namespace = namespace;
+ createSource.name = name;
+ createSource.destinationTopicName = topicName;
+ createSource.deserializationClassName = serdeClassName;
+ createSource.processingGuarantees = processingGuarantees;
+ createSource.parallelism = parallelism;
+ createSource.archive = jarFile;
+ createSource.cpu = cpu;
+ createSource.ram = ram;
+ createSource.disk = disk;
+ createSource.sourceConfigString = sourceConfigString;
+
+ createSource.processArguments();
+
+ createSource.runCmd();
+
+ // test update source
+ updateSource.tenant = tenant;
+ updateSource.namespace = namespace;
+ updateSource.name = name;
+ updateSource.destinationTopicName = topicName;
+ updateSource.deserializationClassName = serdeClassName;
+ updateSource.processingGuarantees = processingGuarantees;
+ updateSource.parallelism = parallelism;
+ updateSource.archive = jarFile;
+ updateSource.cpu = cpu;
+ updateSource.ram = ram;
+ updateSource.disk = disk;
+ updateSource.sourceConfigString = sourceConfigString;
+
+ updateSource.processArguments();
+
+ updateSource.runCmd();
+
+ // test local runner
+ localSourceRunner.tenant = tenant;
+ localSourceRunner.namespace = namespace;
+ localSourceRunner.name = name;
+ localSourceRunner.destinationTopicName = topicName;
+ localSourceRunner.deserializationClassName = serdeClassName;
+ localSourceRunner.processingGuarantees = processingGuarantees;
+ localSourceRunner.parallelism = parallelism;
+ localSourceRunner.archive = jarFile;
+ localSourceRunner.cpu = cpu;
+ localSourceRunner.ram = ram;
+ localSourceRunner.disk = disk;
+ localSourceRunner.sourceConfigString = sourceConfigString;
+
+ localSourceRunner.processArguments();
+
+ localSourceRunner.runCmd();
+
+ verify(createSource).validateSourceConfigs(eq(sourceConfig));
+ verify(updateSource).validateSourceConfigs(eq(sourceConfig));
+ verify(localSourceRunner).validateSourceConfigs(eq(sourceConfig));
+ }
+
+
+ @Test
+ public void testCmdSourceConfigFileCorrect() throws Exception {
+ SourceConfig sourceConfig = getSourceConfig();
+ testCmdSourceConfigFile(sourceConfig, sourceConfig);
+ }
+
+ @Test
+ public void testCmdSourceConfigFileMissingTenant() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setTenant(null);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setTenant(PUBLIC_TENANT);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test
+ public void testCmdSourceConfigFileMissingNamespace() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setNamespace(null);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setNamespace(DEFAULT_NAMESPACE);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Field 'name' cannot be null!")
+ public void testCmdSourceConfigFileMissingName() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setName(null);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setName(null);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Field 'topicName' cannot be null!")
+ public void testCmdSourceConfigFileMissingTopicName() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setTopicName(null);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setTopicName(null);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test
+ public void testCmdSourceConfigFileMissingSerdeClassname() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setSerdeClassName(null);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setSerdeClassName(null);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test
+ public void testCmdSourceConfigFileMissingConfig() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setConfigs(null);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setConfigs(null);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Field 'parallelism' must be a Positive Number")
+ public void testCmdSourceConfigFileZeroParallelism() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setParallelism(0);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setParallelism(0);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Field 'parallelism' must be a Positive Number")
+ public void testCmdSourceConfigFileNegativeParallelism() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setParallelism(-1);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setParallelism(-1);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test
+ public void testCmdSourceConfigFileMissingProcessingGuarantees() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setProcessingGuarantees(null);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setProcessingGuarantees(null);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test
+ public void testCmdSourceConfigFileMissingResources() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setResources(null);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setResources(new Resources());
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Source archive not specfied")
+ public void testCmdSourceConfigFileMissingJar() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setArchive(null);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setArchive(null);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Archive file /tmp/foo.jar does not exist")
+ public void testCmdSourceConfigFileInvalidJar() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setArchive("/tmp/foo.jar");
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setArchive("/tmp/foo.jar");
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "Connector from .*.pulsar-io-cassandra.nar has error: The 'cassandra' connector does not provide a source implementation")
+ public void testCmdSourceConfigFileInvalidJarNoSource() throws Exception {
+ SourceConfig testSourceConfig = getSourceConfig();
+ testSourceConfig.setArchive(WRONG_JAR_PATH);
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+ expectedSourceConfig.setArchive(WRONG_JAR_PATH);
+ testCmdSourceConfigFile(testSourceConfig, expectedSourceConfig);
+ }
+
+ public void testCmdSourceConfigFile(SourceConfig testSourceConfig, SourceConfig expectedSourceConfig) throws Exception {
+
+ File file = Files.createTempFile("", "").toFile();
+
+ new YAMLMapper().writeValue(file, testSourceConfig);
+
+ Assert.assertEquals(testSourceConfig, CmdUtils.loadConfig(file.getAbsolutePath(), SourceConfig.class));
+
+ // test create source
+ createSource.sourceConfigFile = file.getAbsolutePath();
+
+ createSource.processArguments();
+ createSource.runCmd();
+
+ // test update source
+ updateSource.sourceConfigFile = file.getAbsolutePath();
+
+ updateSource.processArguments();
+ updateSource.runCmd();
+
+ // test local runner
+ localSourceRunner.sourceConfigFile = file.getAbsolutePath();
+
+ localSourceRunner.processArguments();
+ localSourceRunner.runCmd();
+
+ verify(createSource).validateSourceConfigs(eq(expectedSourceConfig));
+ verify(updateSource).validateSourceConfigs(eq(expectedSourceConfig));
+ verify(localSourceRunner).validateSourceConfigs(eq(expectedSourceConfig));
+ }
+
+
+ @Test
+ public void testCliOverwriteConfigFile() throws Exception {
+
+ SourceConfig testSourceConfig = new SourceConfig();
+ testSourceConfig.setTenant(TENANT + "-prime");
+ testSourceConfig.setNamespace(NAMESPACE + "-prime");
+ testSourceConfig.setName(NAME + "-prime");
+ testSourceConfig.setTopicName(TOPIC_NAME + "-prime");
+ testSourceConfig.setSerdeClassName(SERDE_CLASS_NAME + "-prime");
+ testSourceConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE);
+ testSourceConfig.setParallelism(PARALLELISM + 1);
+ testSourceConfig.setArchive(JAR_FILE_PATH + "-prime");
+ testSourceConfig.setResources(new Resources(CPU + 1, RAM + 1, DISK + 1));
+ testSourceConfig.setConfigs(createSource.parseConfigs("{\"created_at-prime\":\"Mon Jul 02 00:33:15 +0000 2018\"}"));
+
+
+ SourceConfig expectedSourceConfig = getSourceConfig();
+
+ File file = Files.createTempFile("", "").toFile();
+
+ new YAMLMapper().writeValue(file, testSourceConfig);
+
+ Assert.assertEquals(testSourceConfig, CmdUtils.loadConfig(file.getAbsolutePath(), SourceConfig.class));
+
+ testMixCliAndConfigFile(
+ TENANT,
+ NAMESPACE,
+ NAME,
+ TOPIC_NAME,
+ SERDE_CLASS_NAME,
+ PROCESSING_GUARANTEES,
+ PARALLELISM,
+ JAR_FILE_PATH,
+ CPU,
+ RAM,
+ DISK,
+ SINK_CONFIG_STRING,
+ file.getAbsolutePath(),
+ expectedSourceConfig
+ );
+ }
+
+ public void testMixCliAndConfigFile(
+ String tenant,
+ String namespace,
+ String name,
+ String topicName,
+ String serdeClassName,
+ FunctionConfig.ProcessingGuarantees processingGuarantees,
+ Integer parallelism,
+ String jarFile,
+ Double cpu,
+ Long ram,
+ Long disk,
+ String sourceConfigString,
+ String sourceConfigFile,
+ SourceConfig sourceConfig
+ ) throws Exception {
+
+
+ // test create source
+ createSource.tenant = tenant;
+ createSource.namespace = namespace;
+ createSource.name = name;
+ createSource.destinationTopicName = topicName;
+ createSource.deserializationClassName = serdeClassName;
+ createSource.processingGuarantees = processingGuarantees;
+ createSource.parallelism = parallelism;
+ createSource.archive = jarFile;
+ createSource.cpu = cpu;
+ createSource.ram = ram;
+ createSource.disk = disk;
+ createSource.sourceConfigString = sourceConfigString;
+ createSource.sourceConfigFile = sourceConfigFile;
+
+ createSource.processArguments();
+
+ createSource.runCmd();
+
+ // test update source
+ updateSource.tenant = tenant;
+ updateSource.namespace = namespace;
+ updateSource.name = name;
+ updateSource.destinationTopicName = topicName;
+ updateSource.deserializationClassName = serdeClassName;
+ updateSource.processingGuarantees = processingGuarantees;
+ updateSource.parallelism = parallelism;
+ updateSource.archive = jarFile;
+ updateSource.cpu = cpu;
+ updateSource.ram = ram;
+ updateSource.disk = disk;
+ updateSource.sourceConfigString = sourceConfigString;
+ updateSource.sourceConfigFile = sourceConfigFile;
+
+
+ updateSource.processArguments();
+
+ updateSource.runCmd();
+
+ // test local runner
+ localSourceRunner.tenant = tenant;
+ localSourceRunner.namespace = namespace;
+ localSourceRunner.name = name;
+ localSourceRunner.destinationTopicName = topicName;
+ localSourceRunner.deserializationClassName = serdeClassName;
+ localSourceRunner.processingGuarantees = processingGuarantees;
+ localSourceRunner.parallelism = parallelism;
+ localSourceRunner.archive = jarFile;
+ localSourceRunner.cpu = cpu;
+ localSourceRunner.ram = ram;
+ localSourceRunner.disk = disk;
+ localSourceRunner.sourceConfigString = sourceConfigString;
+ localSourceRunner.sourceConfigFile = sourceConfigFile;
+
+
+ localSourceRunner.processArguments();
+
+ localSourceRunner.runCmd();
+
+ verify(createSource).validateSourceConfigs(eq(sourceConfig));
+ verify(updateSource).validateSourceConfigs(eq(sourceConfig));
+ verify(localSourceRunner).validateSourceConfigs(eq(sourceConfig));
+ }
+
+ @Test
+ public void testDeleteMissingTenant() throws Exception {
+ deleteSource.tenant = null;
+ deleteSource.namespace = NAMESPACE;
+ deleteSource.name = NAME;
+
+ deleteSource.processArguments();
+
+ deleteSource.runCmd();
+
+ verify(functions).deleteFunction(eq(PUBLIC_TENANT), eq(NAMESPACE), eq(NAME));
+ }
+
+ @Test
+ public void testDeleteMissingNamespace() throws Exception {
+ deleteSource.tenant = TENANT;
+ deleteSource.namespace = null;
+ deleteSource.name = NAME;
+
+ deleteSource.processArguments();
+
+ deleteSource.runCmd();
+
+ verify(functions).deleteFunction(eq(TENANT), eq(DEFAULT_NAMESPACE), eq(NAME));
+ }
+
+ @Test(expectedExceptions = ParameterException.class, expectedExceptionsMessageRegExp = "You must specify a name for the source")
+ public void testDeleteMissingName() throws Exception {
+ deleteSource.tenant = TENANT;
+ deleteSource.namespace = NAMESPACE;
+ deleteSource.name = null;
+
+ deleteSource.processArguments();
+
+ deleteSource.runCmd();
+
+ verify(functions).deleteFunction(eq(TENANT), eq(NAMESPACE), null);
+ }
+}
\ No newline at end of file
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 afbd0323bf..2ffbcc945e 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
@@ -54,7 +54,7 @@
private String topicName;
@isImplementationOfClass(implementsClass = SerDe.class)
private String serdeClassName;
- private Map<String, Object> configs = new HashMap<>();
+ private Map<String, Object> configs;
@isPositiveNumber
private int parallelism = 1;
private FunctionConfig.ProcessingGuarantees processingGuarantees;
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services