diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml index 1a24683ae4b87..3bcdd8260e341 100644 --- a/pulsar-broker/pom.xml +++ b/pulsar-broker/pom.xml @@ -111,7 +111,7 @@ pulsar-transaction-common ${project.version} - + ${project.groupId} pulsar-io-batch-discovery-triggerers @@ -145,7 +145,7 @@ io.dropwizard.metrics metrics-core - + org.xerial.snappy @@ -158,7 +158,7 @@ ${project.version} test-jar test - + @@ -315,6 +315,7 @@ ${project.groupId} pulsar-functions-api-examples ${project.version} + pom test @@ -322,6 +323,7 @@ ${project.groupId} pulsar-io-batch-data-generator ${project.version} + pom test @@ -329,6 +331,7 @@ ${project.groupId} pulsar-io-data-generator ${project.version} + pom test @@ -407,49 +410,79 @@ - org.xolstice.maven.plugins - protobuf-maven-plugin - ${protobuf-maven-plugin.version} - - com.google.protobuf:protoc:${protoc3.version}:exe:${os.detected.classifier} - true - + maven-dependency-plugin - generate-sources + copy-pulsar-io-connectors + generate-test-resources - compile - test-compile + copy + + + + ${project.groupId} + pulsar-io-data-generator + ${project.version} + jar + true + ${project.build.directory} + pulsar-io-data-generator.nar + + + ${project.groupId} + pulsar-io-batch-data-generator + ${project.version} + jar + true + ${project.build.directory} + pulsar-io-batch-data-generator.nar + + + ${project.groupId} + pulsar-functions-api-examples + ${project.version} + jar + true + ${project.build.directory} + pulsar-functions-api-examples.jar + + + org.apache.maven.plugins - maven-antrun-plugin + maven-surefire-plugin + + + ${project.build.directory}/pulsar-io-data-generator.nar + ${project.build.directory}/pulsar-functions-api-examples.jar + ${project.build.directory}/pulsar-io-batch-data-generator.nar + + + + + + org.xolstice.maven.plugins + protobuf-maven-plugin + ${protobuf-maven-plugin.version} + + com.google.protobuf:protoc:${protoc3.version}:exe:${os.detected.classifier} + true + - compile + generate-sources - run + compile + test-compile - - - copy test examples package - - - - - - - diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java index 3dea0babf473e..f4260d46b42c5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java @@ -148,6 +148,7 @@ public void initNamespace() throws Exception { @Override @BeforeMethod public void setup() throws Exception { + resetConfig(); conf.setClusterName(testLocalCluster); super.internalSetup(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiTest.java index 2ab8c9d187d4d..9372263d4a7d2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiTest.java @@ -63,8 +63,7 @@ public void testPackagesOperations() throws Exception { admin.packages().upload(originalMetadata, packageName, file.getPath()); // testing download api - String directory = file.getParent(); - String downloadPath = directory + "package-api-test-download.package"; + String downloadPath = new File(file.getParentFile(), "package-api-test-download.package").getPath(); admin.packages().download(packageName, downloadPath); File downloadFile = new File(downloadPath); assertTrue(downloadFile.exists()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/FileServer.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/FileServer.java new file mode 100644 index 0000000000000..6d1fbdb3c3898 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/FileServer.java @@ -0,0 +1,76 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.functions.worker; + +import com.google.api.Http; +import com.sun.net.httpserver.Headers; +import com.sun.net.httpserver.HttpServer; +import java.io.File; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.file.Files; +import lombok.extern.slf4j.Slf4j; + +/** + * Simple http server for serving files in Pulsar Function test cases + */ +@Slf4j +public class FileServer implements AutoCloseable { + private final HttpServer httpServer; + + public FileServer() throws IOException { + httpServer = HttpServer.create(new InetSocketAddress(0), 0); + // creates a default executor + httpServer.setExecutor(null); + } + + public void serveFile(String path, File file) { + httpServer.createContext(path, he -> { + try { + Headers headers = he.getResponseHeaders(); + headers.add("Content-Type", "application/octet-stream"); + + he.sendResponseHeaders(200, file.length()); + try (OutputStream outputStream = he.getResponseBody()) { + Files.copy(file.toPath(), outputStream); + } + } catch (Exception e) { + log.error("Error serving file {} for path {}", file, path, e); + } + }); + } + + public void start() { + httpServer.start(); + } + + public void stop() { + httpServer.stop(0); + } + + public String getUrl(String path) { + return "http://127.0.0.1:" + httpServer.getAddress().getPort() + path; + } + + @Override + public void close() throws Exception { + stop(); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java index bbfdccf4c9533..3b0628b75b025 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java @@ -21,6 +21,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank; import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically; import static org.apache.pulsar.functions.utils.functioncache.FunctionCacheEntry.JAVA_INSTANCE_JAR_PROPERTY; +import static org.apache.pulsar.functions.worker.PulsarFunctionLocalRunTest.getPulsarApiExamplesJar; import static org.mockito.Mockito.spy; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotEquals; @@ -293,7 +294,7 @@ public void testAuthorizationWithAnonymousUser() throws Exception { PulsarAdmin.builder().serviceHttpUrl(brokerServiceUrl).build()) ) { - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); FunctionConfig functionConfig = createFunctionConfig(TENANT, NAMESPACE, functionName, sourceTopic, sinkTopic, subscriptionName); @@ -563,7 +564,7 @@ public void testAuthorization() throws Exception { PulsarAdmin.builder().serviceHttpUrl(brokerServiceUrl).authentication(authToken2).build()) ) { - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); FunctionConfig functionConfig = createFunctionConfig(TENANT, NAMESPACE, functionName, sourceTopic, sinkTopic, subscriptionName); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java index 9775f676587b6..7bb15bce1c3d2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java @@ -27,24 +27,25 @@ import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; - -import java.io.BufferedInputStream; +import com.google.common.collect.Lists; +import com.google.common.collect.Sets; import java.io.File; -import java.io.FileInputStream; -import java.io.OutputStream; +import java.io.IOException; import java.lang.reflect.Method; -import java.net.InetSocketAddress; +import java.net.MalformedURLException; import java.net.URL; +import java.net.URLClassLoader; import java.nio.file.Files; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.concurrent.TimeUnit; - +import lombok.Cleanup; import org.apache.commons.io.FileUtils; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; @@ -70,6 +71,7 @@ import org.apache.pulsar.common.functions.Utils; import org.apache.pulsar.common.io.SinkConfig; import org.apache.pulsar.common.io.SourceConfig; +import org.apache.pulsar.common.nar.NarClassLoader; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.SubscriptionStats; import org.apache.pulsar.common.policies.data.TenantInfo; @@ -77,24 +79,19 @@ import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.ObjectMapperFactory; import org.apache.pulsar.functions.LocalRunner; -import org.apache.pulsar.functions.api.examples.pojo.AvroTestObject; import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory; import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactoryConfig; -import org.apache.pulsar.io.datagenerator.DataGeneratorPrintSink; -import org.apache.pulsar.io.datagenerator.DataGeneratorSource; import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.testng.Assert; +import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeClass; import org.testng.annotations.BeforeMethod; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; -import com.google.common.collect.Lists; -import com.google.common.collect.Sets; -import com.sun.net.httpserver.Headers; -import com.sun.net.httpserver.HttpServer; - /** * Test Pulsar sink on function * @@ -122,14 +119,60 @@ public class PulsarFunctionLocalRunTest { private final String TLS_CLIENT_KEY_FILE_PATH = "./src/test/resources/authentication/tls/client-key.pem"; private final String TLS_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/cacert.pem"; + private static final String SYSTEM_PROPERTY_NAME_NAR_FILE_PATH = "pulsar-io-data-generator.nar.path"; + + public static File getPulsarIODataGeneratorNar() { + return new File(Objects.requireNonNull(System.getProperty(SYSTEM_PROPERTY_NAME_NAR_FILE_PATH) + , "pulsar-io-data-generator.nar file location must be specified with " + + SYSTEM_PROPERTY_NAME_NAR_FILE_PATH + " system property")); + } + + private static final String SYSTEM_PROPERTY_NAME_FUNCTIONS_API_EXAMPLES_JAR_FILE_PATH = + "pulsar-functions-api-examples.jar.path"; + + public static File getPulsarApiExamplesJar() { + return new File(Objects.requireNonNull( + System.getProperty(SYSTEM_PROPERTY_NAME_FUNCTIONS_API_EXAMPLES_JAR_FILE_PATH) + , "pulsar-functions-api-examples.jar file location must be specified with " + + SYSTEM_PROPERTY_NAME_FUNCTIONS_API_EXAMPLES_JAR_FILE_PATH + " system property")); + } + + private static final String SYSTEM_PROPERTY_NAME_BATCH_NAR_FILE_PATH = "pulsar-io-batch-data-generator.nar.path"; + + public static File getPulsarIOBatchDataGeneratorNar() { + return new File(Objects.requireNonNull(System.getProperty(SYSTEM_PROPERTY_NAME_BATCH_NAR_FILE_PATH) + , "pulsar-io-batch-data-generator.nar file location must be specified with " + + SYSTEM_PROPERTY_NAME_BATCH_NAR_FILE_PATH + " system property")); + } + + + private URLClassLoader pulsarApiExamplesClassLoader; + private Class avroTestObjectClass; + + private static final Logger log = LoggerFactory.getLogger(PulsarFunctionLocalRunTest.class); - private HttpServer fileServer; + private FileServer fileServer; @DataProvider(name = "validRoleName") public Object[][] validRoleName() { return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } }; } + @BeforeClass + void loadPulsarApiExamples() throws MalformedURLException, ClassNotFoundException { + pulsarApiExamplesClassLoader = new URLClassLoader(new URL[]{getPulsarApiExamplesJar().toURI().toURL()}, + Thread.currentThread().getContextClassLoader()); + avroTestObjectClass = pulsarApiExamplesClassLoader.loadClass("org.apache.pulsar.functions.api.examples.pojo.AvroTestObject"); + } + + @AfterClass(alwaysRun = true) + void closeClassLoader() throws IOException { + if (pulsarApiExamplesClassLoader != null) { + pulsarApiExamplesClassLoader.close(); + pulsarApiExamplesClassLoader = null; + } + } + @BeforeMethod void setup(Method method) throws Exception { @@ -190,7 +233,7 @@ void setup(Method method) throws Exception { } if (connectorsDir.mkdir()) { - File file = new File(getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile()); + File file = getPulsarIODataGeneratorNar(); Files.copy(file.toPath(), new File(connectorsDir.getAbsolutePath() + "/" + file.getName()).toPath()); } else { throw new RuntimeException("Failed to create builtin connectors directory"); @@ -239,62 +282,16 @@ && isNotBlank(workerConfig.getBrokerClientAuthenticationParameters())) { admin.tenants().createTenant(tenant, propAdmin); // setting up simple web sever to test submitting function via URL - fileServer = HttpServer.create(new InetSocketAddress(0), 0); - fileServer.createContext("/pulsar-io-data-generator.nar", he -> { - try { - - Headers headers = he.getResponseHeaders(); - headers.add("Content-Type", "application/octet-stream"); - - File file = new File(getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile()); - byte[] bytes = new byte[(int) file.length()]; - - FileInputStream fileInputStream = new FileInputStream(file); - BufferedInputStream bufferedInputStream = new BufferedInputStream(fileInputStream); - bufferedInputStream.read(bytes, 0, bytes.length); - - he.sendResponseHeaders(200, file.length()); - OutputStream outputStream = he.getResponseBody(); - outputStream.write(bytes, 0, bytes.length); - outputStream.close(); - - } catch (Exception e) { - log.error("Error when downloading: {}", e, e); - } - }); - fileServer.createContext("/pulsar-functions-api-examples.jar", he -> { - try { - - Headers headers = he.getResponseHeaders(); - headers.add("Content-Type", "application/octet-stream"); - - File file = new File( - getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile()); - byte[] bytes = new byte[(int) file.length()]; - - FileInputStream fileInputStream = new FileInputStream(file); - BufferedInputStream bufferedInputStream = new BufferedInputStream(fileInputStream); - bufferedInputStream.read(bytes, 0, bytes.length); - - he.sendResponseHeaders(200, file.length()); - OutputStream outputStream = he.getResponseBody(); - outputStream.write(bytes, 0, bytes.length); - outputStream.close(); - - } catch (Exception e) { - log.error("Error when downloading: {}", e, e); - } - }); - fileServer.setExecutor(null); // creates a default executor - log.info("Starting file server..."); + fileServer = new FileServer(); + fileServer.serveFile("/pulsar-io-data-generator.nar", getPulsarIODataGeneratorNar()); + fileServer.serveFile("/pulsar-functions-api-examples.jar", getPulsarApiExamplesJar()); fileServer.start(); - } @AfterMethod(alwaysRun = true) void shutdown() throws Exception { log.info("--- Shutting down ---"); - fileServer.stop(0); + fileServer.stop(); pulsarClient.close(); admin.close(); pulsar.close(); @@ -415,6 +412,7 @@ private void testE2EPulsarFunctionLocalRun(String jarFilePathUrl) throws Excepti functionConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE); functionConfig.setJar(jarFilePathUrl); + @Cleanup LocalRunner localRunner = LocalRunner.builder() .functionConfig(functionConfig) .clientAuthPlugin(AuthenticationTls.class.getName()) @@ -521,13 +519,13 @@ public void testAvroFunctionLocalRun(String jarFilePathUrl) throws Exception { Schema schema = Schema.AVRO(SchemaDefinition.builder() .withAlwaysAllowNull(true) .withJSR310ConversionEnabled(true) - .withPojo(AvroTestObject.class).build()); + .withPojo(avroTestObjectClass).build()); //use AVRO schema admin.schemas().createSchema(sourceTopic, schema.getSchemaInfo()); // please note that in this test the sink topic schema is different from the schema of the source topic //produce message to sourceTopic - Producer producer = pulsarClient.newProducer(schema).topic(sourceTopic).create(); + Producer producer = pulsarClient.newProducer(schema).topic(sourceTopic).create(); //consume message from sinkTopic Consumer consumer = pulsarClient.newConsumer(Schema.AUTO_CONSUME()).topic(sinkTopic).subscriptionName("sub").subscribe(); @@ -548,6 +546,7 @@ public void testAvroFunctionLocalRun(String jarFilePathUrl) throws Exception { functionConfig.setJar(jarFilePathUrl); } + @Cleanup LocalRunner localRunner = LocalRunner.builder() .functionConfig(functionConfig) .clientAuthPlugin(AuthenticationTls.class.getName()) @@ -570,9 +569,10 @@ public void testAvroFunctionLocalRun(String jarFilePathUrl) throws Exception { }, 50, 150); int totalMsgs = 5; + Method setBaseValueMethod = avroTestObjectClass.getMethod("setBaseValue", new Class[]{int.class}); for (int i = 0; i < totalMsgs; i++) { - AvroTestObject avroTestObject = new AvroTestObject(); - avroTestObject.setBaseValue(i); + Object avroTestObject = avroTestObjectClass.newInstance(); + setBaseValueMethod.invoke(avroTestObject, i); producer.newMessage().property(propertyKey, propertyValue) .value(avroTestObject).send(); } @@ -613,7 +613,7 @@ public void testAvroFunctionLocalRun(String jarFilePathUrl) throws Exception { .brokerServiceUrl(pulsar.getBrokerServiceUrlTls()).build(); localRunner.start(false); - producer.newMessage().property(propertyKey, propertyValue).value(new AvroTestObject()).send(); + producer.newMessage().property(propertyKey, propertyValue).value(avroTestObjectClass.newInstance()).send(); Message msg = consumer.receive(2, TimeUnit.SECONDS); assertEquals(msg, null); @@ -623,25 +623,24 @@ public void testAvroFunctionLocalRun(String jarFilePathUrl) throws Exception { } @Test(timeOut = 20000) - public void testE2EPulsarFunctionLocalRun() throws Exception { - testE2EPulsarFunctionLocalRun(null); + public void testE2EPulsarFunctionLocalRun() throws Throwable { + runWithPulsarFunctionsClassLoader(() -> testE2EPulsarFunctionLocalRun(null)); } @Test(timeOut = 30000) - public void testAvroFunctionLocalRun() throws Exception { - testAvroFunctionLocalRun(null); + public void testAvroFunctionLocalRun() throws Throwable { + runWithPulsarFunctionsClassLoader(() -> testAvroFunctionLocalRun(null)); } @Test(timeOut = 20000) public void testE2EPulsarFunctionLocalRunWithJar() throws Exception { - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); testE2EPulsarFunctionLocalRun(jarFilePathUrl); } @Test(timeOut = 40000) public void testE2EPulsarFunctionLocalRunURL() throws Exception { - String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-functions-api-examples.jar", fileServer.getAddress().getPort()); - testE2EPulsarFunctionLocalRun(jarFilePathUrl); + testE2EPulsarFunctionLocalRun(fileServer.getUrl("/pulsar-functions-api-examples.jar")); } private void testPulsarSourceLocalRun(String jarFilePathUrl) throws Exception { @@ -655,10 +654,11 @@ private void testPulsarSourceLocalRun(String jarFilePathUrl) throws Exception { SourceConfig sourceConfig = createSourceConfig(tenant, namespacePortion, sourceName, sinkTopic); if (jarFilePathUrl == null || !jarFilePathUrl.endsWith(".nar")) { - sourceConfig.setClassName(DataGeneratorSource.class.getName()); + sourceConfig.setClassName("org.apache.pulsar.io.datagenerator.DataGeneratorSource"); } sourceConfig.setArchive(jarFilePathUrl); + @Cleanup LocalRunner localRunner = LocalRunner.builder() .sourceConfig(sourceConfig) .clientAuthPlugin(AuthenticationTls.class.getName()) @@ -732,20 +732,19 @@ public void testPulsarSourceStatsBuiltin() throws Exception { } @Test(timeOut = 20000) - public void testPulsarSourceLocalRunNoArchive() throws Exception { - testPulsarSourceLocalRun(null); + public void testPulsarSourceLocalRunNoArchive() throws Throwable { + runWithNarClassLoader(() -> testPulsarSourceLocalRun(null)); } @Test(timeOut = 20000) public void testPulsarSourceLocalRunWithFile() throws Exception { - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile(); + String jarFilePathUrl = getPulsarIODataGeneratorNar().toURI().toString(); testPulsarSourceLocalRun(jarFilePathUrl); } @Test(timeOut = 40000) public void testPulsarSourceLocalRunWithUrl() throws Exception { - String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-io-data-generator.nar", fileServer.getAddress().getPort()); - testPulsarSourceLocalRun(jarFilePathUrl); + testPulsarSourceLocalRun(fileServer.getUrl("/pulsar-io-data-generator.nar")); } @@ -768,10 +767,11 @@ private void testPulsarSinkLocalRun(String jarFilePathUrl) throws Exception { sinkConfig.setInputSpecs(Collections.singletonMap(sourceTopic, ConsumerConfig.builder().receiverQueueSize(1000).build())); if (jarFilePathUrl == null || !jarFilePathUrl.endsWith(".nar")) { - sinkConfig.setClassName(DataGeneratorPrintSink.class.getName()); + sinkConfig.setClassName("org.apache.pulsar.io.datagenerator.DataGeneratorPrintSink"); } sinkConfig.setArchive(jarFilePathUrl); + @Cleanup LocalRunner localRunner = LocalRunner.builder() .sinkConfig(sinkConfig) .clientAuthPlugin(AuthenticationTls.class.getName()) @@ -842,19 +842,40 @@ public void testPulsarSinkStatsBuiltin() throws Exception { } @Test(timeOut = 20000) - public void testPulsarSinkStatsNoArchive() throws Exception { - testPulsarSinkLocalRun(null); + public void testPulsarSinkStatsNoArchive() throws Throwable { + runWithNarClassLoader(() -> testPulsarSinkLocalRun(null)); + } + + private void runWithNarClassLoader(Assert.ThrowingRunnable throwingRunnable) throws Throwable { + ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader(); + try (NarClassLoader classLoader = NarClassLoader.getFromArchive(getPulsarIODataGeneratorNar(), Collections.emptySet(), originalClassLoader, NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR)) { + try { + Thread.currentThread().setContextClassLoader(classLoader); + throwingRunnable.run(); + } finally { + Thread.currentThread().setContextClassLoader(originalClassLoader); + } + } + } + + private void runWithPulsarFunctionsClassLoader(Assert.ThrowingRunnable throwingRunnable) throws Throwable { + ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader(); + try { + Thread.currentThread().setContextClassLoader(pulsarApiExamplesClassLoader); + throwingRunnable.run(); + } finally { + Thread.currentThread().setContextClassLoader(originalClassLoader); + } } @Test(timeOut = 20000) public void testPulsarSinkStatsWithFile() throws Exception { - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile(); + String jarFilePathUrl = getPulsarIODataGeneratorNar().toURI().toString(); testPulsarSinkLocalRun(jarFilePathUrl); } @Test(timeOut = 40000) public void testPulsarSinkStatsWithUrl() throws Exception { - String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-io-data-generator.nar", fileServer.getAddress().getPort()); - testPulsarSinkLocalRun(jarFilePathUrl); + testPulsarSinkLocalRun(fileServer.getUrl("/pulsar-io-data-generator.nar")); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java index 77a7aab34c46a..ce2cfaf62c0a3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java @@ -21,6 +21,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank; import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically; import static org.apache.pulsar.functions.utils.functioncache.FunctionCacheEntry.JAVA_INSTANCE_JAR_PROPERTY; +import static org.apache.pulsar.functions.worker.PulsarFunctionLocalRunTest.getPulsarApiExamplesJar; import static org.mockito.Mockito.spy; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotEquals; @@ -302,7 +303,7 @@ public void testPulsarFunctionState() throws Exception { FunctionConfig functionConfig = createFunctionConfig(tenant, namespacePortion, functionName, sourceTopic, publishTopic, subscriptionName); - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); admin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl); retryStrategically((test) -> { @@ -400,10 +401,9 @@ public void testMultipleAddress() throws Exception { .allowTlsInsecureConnection(true).authentication(authTls) .build(); - String jarFilePath = getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); - File jarFile = new File(jarFilePath); + File jarFile = getPulsarApiExamplesJar(); Assert.assertTrue(jarFile.exists() && jarFile.isFile()); - pulsarAdmin.functions().createFunction(functionConfig, jarFilePath); + pulsarAdmin.functions().createFunction(functionConfig, jarFile.getAbsolutePath()); retryStrategically((test) -> { try { return admin.topics().getStats(sourceTopic).subscriptions.size() == 1; @@ -443,10 +443,9 @@ public void testPulsarFunctionBKCleanup() throws Exception { FunctionConfig functionConfig = createFunctionConfig(tenant, namespacePortion, functionName, sourceTopic, publishTopic, subscriptionName); - String jarFilePath = getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); - File jarFile = new File(jarFilePath); + File jarFile = getPulsarApiExamplesJar(); Assert.assertTrue(jarFile.exists() && jarFile.isFile()); - admin.functions().createFunction(functionConfig, jarFilePath); + admin.functions().createFunction(functionConfig, jarFile.getAbsolutePath()); retryStrategically((test) -> { try { 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 4c57d2e6a95aa..45601aa5df02e 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 @@ -19,6 +19,7 @@ package org.apache.pulsar.functions.worker; import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically; +import static org.apache.pulsar.functions.worker.PulsarFunctionLocalRunTest.getPulsarApiExamplesJar; import static org.mockito.Mockito.spy; import static org.testng.Assert.assertEquals; @@ -178,7 +179,7 @@ public void testFunctionAssignments() throws Exception { final Set clusters = Sets.newHashSet(Lists.newArrayList("use")); admin.namespaces().setNamespaceReplicationClusters(replNamespace, clusters); - final String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + final String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); FunctionConfig functionConfig = createFunctionConfig(tenant, namespacePortion, functionName, "my.*", sinkTopic, subscriptionName); functionConfig.setParallelism(2); @@ -232,7 +233,7 @@ public void testFunctionAssignmentsWithRestart() throws Exception { admin.namespaces().setNamespaceReplicationClusters(replNamespace, clusters); final FunctionRuntimeManager runtimeManager = functionsWorkerService.getFunctionRuntimeManager(); - final String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + final String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); FunctionConfig functionConfig; // (1) Register functions with 2 instances for (int i = 0; i < totalFunctions; i++) { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java index 39d7bb9c644a3..2f4089e177097 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java @@ -22,34 +22,25 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank; import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically; import static org.apache.pulsar.functions.utils.functioncache.FunctionCacheEntry.JAVA_INSTANCE_JAR_PROPERTY; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.eq; +import static org.apache.pulsar.functions.worker.PulsarFunctionLocalRunTest.getPulsarApiExamplesJar; +import static org.apache.pulsar.functions.worker.PulsarFunctionLocalRunTest.getPulsarIOBatchDataGeneratorNar; +import static org.apache.pulsar.functions.worker.PulsarFunctionLocalRunTest.getPulsarIODataGeneratorNar; import static org.mockito.Mockito.spy; -import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotEquals; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; - import com.google.common.collect.Lists; import com.google.common.collect.Sets; import com.google.common.util.concurrent.ThreadFactoryBuilder; -import com.sun.net.httpserver.Headers; -import com.sun.net.httpserver.HttpServer; - -import java.io.BufferedInputStream; import java.io.BufferedReader; import java.io.File; -import java.io.FileInputStream; import java.io.IOException; import java.io.InputStreamReader; -import java.io.OutputStream; import java.lang.reflect.Method; import java.net.HttpURLConnection; -import java.net.InetSocketAddress; import java.net.URL; import java.nio.file.Files; import java.util.Arrays; @@ -60,22 +51,18 @@ import java.util.Optional; import java.util.Set; import java.util.TreeMap; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.regex.Matcher; import java.util.regex.Pattern; - +import lombok.Cleanup; import lombok.ToString; - import org.apache.commons.io.FileUtils; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.ServiceConfigurationUtils; import org.apache.pulsar.broker.authentication.AuthenticationProviderTls; -import org.apache.pulsar.broker.authorization.AuthorizationService; import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider; import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; import org.apache.pulsar.client.admin.BrokerStats; @@ -108,14 +95,16 @@ import org.apache.pulsar.functions.LocalRunner; import org.apache.pulsar.functions.instance.InstanceUtils; import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory; +import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactoryConfig; import org.apache.pulsar.functions.utils.FunctionCommon; +import org.apache.pulsar.functions.worker.FileServer; import org.apache.pulsar.functions.worker.FunctionRuntimeManager; -import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactoryConfig; import org.apache.pulsar.functions.worker.PulsarWorkerService; import org.apache.pulsar.functions.worker.WorkerConfig; import org.apache.pulsar.functions.worker.WorkerService; import org.apache.pulsar.io.batchdiscovery.ImmediateTriggerer; import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble; +import org.awaitility.Awaitility; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testng.Assert; @@ -150,8 +139,7 @@ public class PulsarFunctionE2ETest { private final String TLS_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/cacert.pem"; private static final Logger log = LoggerFactory.getLogger(PulsarFunctionE2ETest.class); - private Thread fileServerThread; - private HttpServer fileServer; + private FileServer fileServer; @DataProvider(name = "validRoleName") public Object[][] validRoleName() { @@ -211,7 +199,7 @@ void setup(Method method) throws Exception { FutureUtil.class.getProtectionDomain().getCodeSource().getLocation().getPath()); functionsWorkerService = createPulsarFunctionWorker(config); - + // populate builtin connectors folder if (Arrays.asList(method.getAnnotation(Test.class).groups()).contains("builtin")) { File connectorsDir = new File(workerConfig.getConnectorsDirectory()); @@ -221,10 +209,10 @@ void setup(Method method) throws Exception { } if (connectorsDir.mkdir()) { - File file = new File(getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile()); + File file = getPulsarIODataGeneratorNar(); Files.copy(file.toPath(), new File(connectorsDir.getAbsolutePath() + "/" + file.getName()).toPath()); - - file = new File(getClass().getClassLoader().getResource("pulsar-io-batch-data-generator.nar").getFile()); + + file = getPulsarIOBatchDataGeneratorNar(); Files.copy(file.toPath(), new File(connectorsDir.getAbsolutePath() + "/" + file.getName()).toPath()); } else { throw new RuntimeException("Failed to create builtin connectors directory"); @@ -268,105 +256,27 @@ && isNotBlank(workerConfig.getBrokerClientAuthenticationParameters())) { propAdmin.setAllowedClusters(Sets.newHashSet(Lists.newArrayList("use"))); admin.tenants().updateTenant(tenant, propAdmin); - // setting up simple web sever to test submitting function via URL - CountDownLatch latch = new CountDownLatch(1); - fileServerThread = new Thread(() -> { - try { - fileServer = HttpServer.create(new InetSocketAddress(0), 0); - fileServer.createContext("/pulsar-io-batch-data-generator.nar", he -> { - try { - - Headers headers = he.getResponseHeaders(); - headers.add("Content-Type", "application/octet-stream"); - - File file = new File(getClass().getClassLoader().getResource("pulsar-io-batch-data-generator.nar").getFile()); - byte[] bytes = new byte [(int)file.length()]; - - FileInputStream fileInputStream = new FileInputStream(file); - BufferedInputStream bufferedInputStream = new BufferedInputStream(fileInputStream); - bufferedInputStream.read(bytes, 0, bytes.length); - bufferedInputStream.close(); - fileInputStream.close(); - - he.sendResponseHeaders(200, file.length()); - OutputStream outputStream = he.getResponseBody(); - outputStream.write(bytes, 0, bytes.length); - outputStream.close(); - - } catch (Exception e) { - log.error("Error when downloading: {}", e, e); - } - }); - fileServer.createContext("/pulsar-io-data-generator.nar", he -> { - try { - - Headers headers = he.getResponseHeaders(); - headers.add("Content-Type", "application/octet-stream"); - - File file = new File(getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile()); - byte[] bytes = new byte [(int)file.length()]; - - FileInputStream fileInputStream = new FileInputStream(file); - BufferedInputStream bufferedInputStream = new BufferedInputStream(fileInputStream); - bufferedInputStream.read(bytes, 0, bytes.length); - bufferedInputStream.close(); - fileInputStream.close(); - - he.sendResponseHeaders(200, file.length()); - OutputStream outputStream = he.getResponseBody(); - outputStream.write(bytes, 0, bytes.length); - outputStream.close(); - - } catch (Exception e) { - log.error("Error when downloading: {}", e, e); - } - }); - fileServer.createContext("/pulsar-functions-api-examples.jar", he -> { - try { - - Headers headers = he.getResponseHeaders(); - headers.add("Content-Type", "application/octet-stream"); - - File file = new File(getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile()); - byte[] bytes = new byte [(int)file.length()]; - - FileInputStream fileInputStream = new FileInputStream(file); - BufferedInputStream bufferedInputStream = new BufferedInputStream(fileInputStream); - bufferedInputStream.read(bytes, 0, bytes.length); - bufferedInputStream.close(); - fileInputStream.close(); - - he.sendResponseHeaders(200, file.length()); - OutputStream outputStream = he.getResponseBody(); - outputStream.write(bytes, 0, bytes.length); - outputStream.close(); - - } catch (Exception e) { - log.error("Error when downloading: {}", e, e); - } - }); - fileServer.setExecutor(null); // creates a default executor - log.info("Starting file server..."); - fileServer.start(); - } catch (Exception e) { - log.error("Failed to start file server: ", e); - fileServer.stop(0); - } - latch.countDown(); - }); - fileServerThread.start(); - latch.await(1, TimeUnit.SECONDS); - - while (!functionsWorkerService.getLeaderService().isLeader()) { - Thread.sleep(1000); - } + assertTrue(getPulsarIODataGeneratorNar().exists(), "pulsar-io-data-generator.nar file " + + getPulsarIODataGeneratorNar().getAbsolutePath() + " doesn't exist."); + assertTrue(getPulsarIOBatchDataGeneratorNar().exists(), "pulsar-io-batch-data-generator.nar file " + + getPulsarIOBatchDataGeneratorNar().getAbsolutePath() + " doesn't exist."); + assertTrue(getPulsarApiExamplesJar().exists(), "pulsar-functions-api-examples.jar file " + + getPulsarApiExamplesJar().getAbsolutePath() + " doesn't exist."); + + // setting up simple web server to test submitting function via URL + fileServer = new FileServer(); + fileServer.serveFile("/pulsar-io-data-generator.nar", getPulsarIODataGeneratorNar()); + fileServer.serveFile("/pulsar-io-batch-data-generator.nar", getPulsarIOBatchDataGeneratorNar()); + fileServer.serveFile("/pulsar-functions-api-examples.jar", getPulsarApiExamplesJar()); + fileServer.start(); + + Awaitility.await().until(() -> functionsWorkerService.getLeaderService().isLeader()); } @AfterMethod(alwaysRun = true) void shutdown() throws Exception { log.info("--- Shutting down ---"); - fileServer.stop(0); - fileServerThread.interrupt(); + fileServer.stop(); pulsarClient.close(); admin.close(); functionsWorkerService.stop(); @@ -378,7 +288,7 @@ void shutdown() throws Exception { FileUtils.deleteDirectory(connectorsDir); } } - + private PulsarWorkerService createPulsarFunctionWorker(ServiceConfiguration config) throws IOException { System.setProperty(JAVA_INSTANCE_JAR_PROPERTY, @@ -463,13 +373,13 @@ private static SinkConfig createSinkConfig(String tenant, String namespace, Stri sinkConfig.setCleanupSubscription(true); return sinkConfig; } - + private static BatchSourceConfig createBatchSourceConfig() { return BatchSourceConfig.builder() .discoveryTriggererClassName(ImmediateTriggerer.class.getName()) .build(); } - + /** * Validates pulsar sink e2e functionality on functions. * @@ -504,47 +414,30 @@ private void testE2EPulsarFunction(String jarFilePathUrl) throws Exception { functionConfig.setOutput(sinkTopic2); admin.functions().updateFunctionWithUrl(functionConfig, jarFilePathUrl); - retryStrategically((test) -> { - try { - TopicStats topicStats = admin.topics().getStats(sinkTopic2); - return topicStats.publishers.size() == 2 - && topicStats.publishers.get(0).metadata != null - && topicStats.publishers.get(0).metadata.containsKey("id") - && topicStats.publishers.get(0).metadata.get("id").equals(String.format("%s/%s/%s", tenant, namespacePortion, functionName)); - } catch (PulsarAdminException e) { - return false; - } - }, 50, 150); - - TopicStats topicStats = admin.topics().getStats(sinkTopic2); - assertEquals(topicStats.publishers.size(), 2); - assertNotNull(topicStats.publishers.get(0).metadata); - assertTrue(topicStats.publishers.get(0).metadata.containsKey("id")); - assertEquals(topicStats.publishers.get(0).metadata.get("id"), String.format("%s/%s/%s", tenant, namespacePortion, functionName)); + Awaitility.await().ignoreExceptions().untilAsserted(() -> { + TopicStats topicStats = admin.topics().getStats(sinkTopic2); + assertEquals(topicStats.publishers.size(), 2); + assertNotNull(topicStats.publishers.get(0).metadata); + assertTrue(topicStats.publishers.get(0).metadata.containsKey("id")); + assertEquals(topicStats.publishers.get(0).metadata.get("id"), + String.format("%s/%s/%s", tenant, namespacePortion, functionName)); + }); - retryStrategically((test) -> { - try { - return admin.topics().getStats(sourceTopic).subscriptions.size() == 1; - } catch (PulsarAdminException e) { - return false; - } - }, 50, 150); - // validate pulsar sink consumer has started on the topic - assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1); + Awaitility.await().ignoreExceptions().untilAsserted(() -> { + // validate pulsar sink consumer has started on the topic + assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1); + }); int totalMsgs = 5; for (int i = 0; i < totalMsgs; i++) { String data = "my-message-" + i; producer.newMessage().property(propertyKey, propertyValue).value(data).send(); } - retryStrategically((test) -> { - try { - SubscriptionStats subStats = admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName); - return subStats.unackedMessages == 0; - } catch (PulsarAdminException e) { - return false; - } - }, 50, 150); + + Awaitility.await().ignoreExceptions().untilAsserted(() -> { + SubscriptionStats subStats = admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName); + assertEquals(subStats.unackedMessages, 0); + }); Message msg = consumer.receive(5, TimeUnit.SECONDS); String receivedPropertyValue = msg.getProperty(propertyKey); @@ -558,16 +451,10 @@ private void testE2EPulsarFunction(String jarFilePathUrl) throws Exception { // delete functions admin.functions().deleteFunction(tenant, namespacePortion, functionName); - retryStrategically((test) -> { - try { - return admin.topics().getStats(sourceTopic).subscriptions.size() == 0; - } catch (PulsarAdminException e) { - return false; - } - }, 50, 150); - - // make sure subscriptions are cleanup - assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 0); + Awaitility.await().ignoreExceptions().untilAsserted(() -> { + // make sure subscriptions are cleanup + assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 0); + }); // make sure all temp files are deleted File dir = new File(System.getProperty("java.io.tmpdir")); @@ -578,15 +465,13 @@ private void testE2EPulsarFunction(String jarFilePathUrl) throws Exception { @Test(timeOut = 20000) public void testE2EPulsarFunctionWithFile() throws Exception { - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); testE2EPulsarFunction(jarFilePathUrl); } @Test(timeOut = 40000) public void testE2EPulsarFunctionWithUrl() throws Exception { - String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-functions-api-examples.jar", - fileServer.getAddress().getPort()); - testE2EPulsarFunction(jarFilePathUrl); + testE2EPulsarFunction(fileServer.getUrl("/pulsar-functions-api-examples.jar")); } @Test(timeOut = 30000) @@ -635,7 +520,7 @@ public void testReadCompactedFunction() throws Exception { consumerConfig.setConsumerProperties(consumerProperties); inputSpecs.put(sourceTopic, consumerConfig); functionConfig.setInputSpecs(inputSpecs); - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); admin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl); // 5 Function should only read compacted value，so we will only receive compacted messages @@ -699,7 +584,7 @@ public void testReadCompactedSink() throws Exception { Map consumerProperties = new HashMap<>(); consumerProperties.put("readCompacted","true"); sinkConfig.setInputSpecs(Collections.singletonMap(sourceTopic, ConsumerConfig.builder().consumerProperties(consumerProperties).build())); - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile(); + String jarFilePathUrl = getPulsarIODataGeneratorNar().toURI().toString(); admin.sink().createSinkWithUrl(sinkConfig, jarFilePathUrl); // 5 Sink should only read compacted value，so we will only receive compacted messages @@ -743,6 +628,7 @@ private void testPulsarSinkDLQ() throws Exception { sinkConfig.setDeadLetterTopic(dlqTopic); sinkConfig.setInputSpecs(Collections.singletonMap(sourceTopic, ConsumerConfig.builder().receiverQueueSize(1000).build())); sinkConfig.setClassName(SinkForTest.class.getName()); + @Cleanup LocalRunner localRunner = LocalRunner.builder() .sinkConfig(sinkConfig) .clientAuthPlugin(AuthenticationTls.class.getName()) @@ -770,17 +656,22 @@ private void testPulsarSinkDLQ() throws Exception { // 3 send message int totalMsgs = 10; + Set remainingMessagesToReceive = new HashSet<>(); for (int i = 0; i < totalMsgs; i++) { - producer.newMessage().property(propertyKey, propertyValue).value("fail" + i).sendAsync(); + String messageBody = "fail" + i; + producer.newMessage().property(propertyKey, propertyValue).value(messageBody).send(); + remainingMessagesToReceive.add(messageBody); } //4 All messages should enter DLQ for (int i = 0; i < totalMsgs; i++) { Message message = consumer.receive(10, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(message.getValue(), "fail" + i); + remainingMessagesToReceive.remove(message.getValue()); } + assertEquals(remainingMessagesToReceive, Collections.emptySet()); + //clean up producer.close(); consumer.close(); @@ -1022,15 +913,13 @@ public void testPulsarSinkStatsBuiltin() throws Exception { @Test(timeOut = 20000) public void testPulsarSinkStatsWithFile() throws Exception { - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile(); + String jarFilePathUrl = getPulsarIODataGeneratorNar().toURI().toString(); testPulsarSinkStats(jarFilePathUrl); } @Test(timeOut = 40000) public void testPulsarSinkStatsWithUrl() throws Exception { - String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-io-data-generator.nar", - fileServer.getAddress().getPort()); - testPulsarSinkStats(jarFilePathUrl); + testPulsarSinkStats(fileServer.getUrl("/pulsar-io-data-generator.nar")); } private void testPulsarSourceStats(String jarFilePathUrl) throws Exception { @@ -1178,17 +1067,15 @@ public void testPulsarSourceStatsBuiltin() throws Exception { @Test(timeOut = 20000) public void testPulsarSourceStatsWithFile() throws Exception { - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile(); + String jarFilePathUrl = getPulsarIODataGeneratorNar().toURI().toString(); testPulsarSourceStats(jarFilePathUrl); } @Test(timeOut = 40000) public void testPulsarSourceStatsWithUrl() throws Exception { - String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-io-data-generator.nar", - fileServer.getAddress().getPort()); - testPulsarSourceStats(jarFilePathUrl); + testPulsarSourceStats(fileServer.getUrl("/pulsar-io-data-generator.nar")); } - + private void testPulsarBatchSourceStats(String jarFilePathUrl) throws Exception { final String namespacePortion = "io"; final String replNamespace = tenant + "/" + namespacePortion; @@ -1200,7 +1087,7 @@ private void testPulsarBatchSourceStats(String jarFilePathUrl) throws Exception SourceConfig sourceConfig = createSourceConfig(tenant, namespacePortion, sourceName, sinkTopic); sourceConfig.setBatchSourceConfig(createBatchSourceConfig()); - + retryStrategically((test) -> { try { return (admin.topics().getStats(sinkTopic).publishers.size() == 1); @@ -1211,12 +1098,12 @@ private void testPulsarBatchSourceStats(String jarFilePathUrl) throws Exception final String sinkTopic2 = "persistent://" + replNamespace + "/output-" + sourceName; sourceConfig.setTopicName(sinkTopic2); - + if (jarFilePathUrl.startsWith(Utils.BUILTIN)) { sourceConfig.setArchive(jarFilePathUrl); admin.sources().createSource(sourceConfig, jarFilePathUrl); } else { - admin.sources().createSourceWithUrl(sourceConfig, jarFilePathUrl); + admin.sources().createSourceWithUrl(sourceConfig, jarFilePathUrl); } retryStrategically((test) -> { @@ -1327,18 +1214,16 @@ public void testPulsarBatchSourceStatsBuiltin() throws Exception { String jarFilePathUrl = String.format("%s://batch-data-generator", Utils.BUILTIN); testPulsarBatchSourceStats(jarFilePathUrl); } - + @Test(timeOut = 20000) private void testPulsarBatchSourceStatsWithFile() throws Exception { - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-io-batch-data-generator.nar").getFile(); + String jarFilePathUrl = getPulsarIOBatchDataGeneratorNar().toURI().toString(); testPulsarBatchSourceStats(jarFilePathUrl); } - + @Test(timeOut = 40000) private void testPulsarBatchSourceStatsWithUrl() throws Exception { - String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-io-batch-data-generator.nar", - fileServer.getAddress().getPort()); - testPulsarBatchSourceStats(jarFilePathUrl); + testPulsarBatchSourceStats(fileServer.getUrl("/pulsar-io-batch-data-generator.nar")); } @Test(timeOut = 20000) @@ -1359,7 +1244,7 @@ public void testPulsarFunctionStats() throws Exception { // create a producer that creates a topic at broker Producer producer = pulsarClient.newProducer(Schema.STRING).topic(sourceTopic).create(); - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); FunctionConfig functionConfig = createFunctionConfig(tenant, namespacePortion, functionName, "my.*", sinkTopic, subscriptionName); admin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl); @@ -1695,7 +1580,7 @@ public void testPulsarFunctionStatus() throws Exception { // create a producer that creates a topic at broker Producer producer = pulsarClient.newProducer(Schema.STRING).topic(sourceTopic).create(); - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); FunctionConfig functionConfig = createFunctionConfig(tenant, namespacePortion, functionName, "my.*", sinkTopic, subscriptionName); admin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl); @@ -1775,8 +1660,7 @@ public void testAuthorization(boolean validRoleName) throws Exception { propAdmin.setAllowedClusters(Sets.newHashSet(Lists.newArrayList("use"))); admin.tenants().updateTenant(tenant, propAdmin); - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader() - .getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); FunctionConfig functionConfig = createFunctionConfig(tenant, namespacePortion, functionName, "my.*", sinkTopic, subscriptionName); if (!validRoleName) { @@ -1818,7 +1702,7 @@ public void testFunctionStopAndRestartApi() throws Exception { // create source topic Producer producer = pulsarClient.newProducer(Schema.STRING).topic(sourceTopic).create(); - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); FunctionConfig functionConfig = createFunctionConfig(tenant, namespacePortion, functionName, sourceTopicName, sinkTopic, subscriptionName); admin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl); @@ -1884,7 +1768,7 @@ public void testFunctionAutomaticSubCleanup() throws Exception { // create a producer that creates a topic at broker Producer producer = pulsarClient.newProducer(Schema.STRING).topic(sourceTopic).create(); - String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + String jarFilePathUrl = getPulsarApiExamplesJar().toURI().toString(); FunctionConfig functionConfig = new FunctionConfig(); functionConfig.setTenant(tenant); functionConfig.setNamespace(namespacePortion); diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java index 095cbf7a97be6..ae84649d7f1bf 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java @@ -160,7 +160,12 @@ public void download(String packageName, String path) throws PulsarAdminExceptio Thread.currentThread().interrupt(); throw new PulsarAdminException(e); } catch (ExecutionException e) { - throw (PulsarAdminException) e.getCause(); + Throwable cause = e.getCause(); + if (cause instanceof PulsarAdminException) { + throw (PulsarAdminException) cause; + } else { + throw new PulsarAdminException(cause); + } } } diff --git a/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java b/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java index 61be4fa9591f3..7d84bd8a1cd27 100644 --- a/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java +++ b/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java @@ -18,19 +18,36 @@ */ package org.apache.pulsar.functions; +import static org.apache.pulsar.common.functions.Utils.inferMissingArguments; import com.beust.jcommander.IStringConverter; import com.beust.jcommander.JCommander; import com.beust.jcommander.Parameter; import com.google.gson.Gson; import com.google.gson.GsonBuilder; import com.google.gson.JsonParser; - +import java.io.Closeable; +import java.io.File; +import java.io.IOException; +import java.nio.file.Paths; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Timer; +import java.util.TimerTask; +import java.util.TreeMap; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import lombok.Builder; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.common.functions.AuthenticationConfig; import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.functions.Utils; import org.apache.pulsar.common.io.SinkConfig; import org.apache.pulsar.common.io.SourceConfig; import org.apache.pulsar.common.nar.NarClassLoader; @@ -57,29 +74,15 @@ import org.apache.pulsar.functions.utils.io.ConnectorUtils; import org.apache.pulsar.functions.worker.WorkerConfig; -import java.io.File; -import java.io.IOException; -import java.nio.file.Paths; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.Timer; -import java.util.TimerTask; -import java.util.TreeMap; -import java.util.UUID; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; - -import static org.apache.pulsar.common.functions.Utils.inferMissingArguments; - @Slf4j -public class LocalRunner { +public class LocalRunner implements AutoCloseable { private final AtomicBoolean running = new AtomicBoolean(false); private final List spawners = new LinkedList<>(); - private String narExtractionDirectory = NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR; + private final String narExtractionDirectory; + private final Thread shutdownHook; + private ClassLoader userCodeClassLoader; + private boolean userCodeClassLoaderCreated; public enum RuntimeEnv { THREAD, @@ -177,7 +180,7 @@ public LocalRunner(FunctionConfig functionConfig, SourceConfig sourceConfig, Sin stateStorageServiceUrl, String brokerServiceUrl, String clientAuthPlugin, String clientAuthParams, boolean useTls, boolean tlsAllowInsecureConnection, boolean tlsHostNameVerificationEnabled, String tlsTrustCertFilePath, int instanceIdOffset, RuntimeEnv runtimeEnv, - String secretsProviderClassName, String secretsProviderConfig) { + String secretsProviderClassName, String secretsProviderConfig, String narExtractionDirectory) { this.functionConfig = functionConfig; this.sourceConfig = sourceConfig; this.sinkConfig = sinkConfig; @@ -193,66 +196,90 @@ public LocalRunner(FunctionConfig functionConfig, SourceConfig sourceConfig, Sin this.runtimeEnv = runtimeEnv; this.secretsProviderClassName = secretsProviderClassName; this.secretsProviderConfig = secretsProviderConfig; - - java.lang.Runtime.getRuntime().addShutdownHook(new Thread() { + this.narExtractionDirectory = narExtractionDirectory != null ? narExtractionDirectory + : NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR; + shutdownHook = new Thread() { public void run() { LocalRunner.this.stop(); } - }); + }; + } + + @Override + public void close() throws Exception { + stop(); } public synchronized void stop() { - running.set(false); - log.info("Shutting down the localrun runtimeSpawner ..."); - for (RuntimeSpawner spawner : spawners) { - spawner.close(); + if (running.compareAndSet(true, false)) { + try { + Runtime.getRuntime().removeShutdownHook(shutdownHook); + } catch (IllegalStateException e) { + // ignore possible "Shutdown in progress" + } + log.info("Shutting down the localrun runtimeSpawner ..."); + for (RuntimeSpawner spawner : spawners) { + spawner.close(); + } + spawners.clear(); + + if (userCodeClassLoaderCreated) { + if (userCodeClassLoader instanceof Closeable) { + try { + ((Closeable) userCodeClassLoader).close(); + } catch (IOException e) { + log.warn("Error closing classloader", e); + } + } + userCodeClassLoaderCreated = false; + userCodeClassLoader = null; + } } - spawners.clear(); } public void start(boolean blocking) throws Exception { List local = new LinkedList<>(); synchronized (this) { - if (running.get() == true) { + if (!running.compareAndSet(false, true)) { throw new IllegalArgumentException("Pulsar Function local run already started!"); } - + Runtime.getRuntime().addShutdownHook(shutdownHook); Function.FunctionDetails functionDetails; - String userCodeFile; + String userCodeFile = null; int parallelism; if (functionConfig != null) { FunctionConfigUtils.inferMissingArguments(functionConfig, true); - ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); parallelism = functionConfig.getParallelism(); if (functionConfig.getRuntime() == FunctionConfig.Runtime.JAVA) { userCodeFile = functionConfig.getJar(); - // if code file not specified try to get location of the code based on class. - if (userCodeFile == null && functionConfig.getClassName() != null) { - userCodeFile = Thread.currentThread().getContextClassLoader() - .loadClass(functionConfig.getClassName()) - .getProtectionDomain().getCodeSource().getLocation().getFile(); - } - - boolean isBuiltin = !org.apache.commons.lang3.StringUtils.isEmpty(functionConfig.getJar()) && functionConfig.getJar().startsWith(org.apache.pulsar.common.functions.Utils.BUILTIN); - if(isBuiltin){ + boolean isBuiltin = !StringUtils.isEmpty(functionConfig.getJar()) + && functionConfig.getJar().startsWith(Utils.BUILTIN); + if (isBuiltin){ WorkerConfig workerConfig = WorkerConfig.load(System.getenv("PULSAR_HOME") + "/conf/functions_worker.yml"); Functions functions = FunctionUtils.searchForFunctions(System.getenv("PULSAR_HOME") + workerConfig.getFunctionsDirectory().replaceFirst("^.", "")); String functionType = functionConfig.getJar().replaceFirst("^builtin://", ""); userCodeFile = functions.getFunctions().get(functionType).toString(); } - if (org.apache.pulsar.common.functions.Utils.isFunctionPackageUrlSupported(userCodeFile)) { + if (Utils.isFunctionPackageUrlSupported(userCodeFile)) { File file = FunctionCommon.extractFileFromPkgURL(userCodeFile); - classLoader = FunctionConfigUtils.validate(functionConfig, file); - } else { + userCodeClassLoader = FunctionConfigUtils.validate(functionConfig, file); + userCodeClassLoaderCreated = true; + } else if (userCodeFile != null) { File file = new File(userCodeFile); if (!file.exists()) { throw new RuntimeException("User jar does not exist"); } - classLoader = FunctionConfigUtils.validate(functionConfig, file); + userCodeClassLoader = FunctionConfigUtils.validate(functionConfig, file); + userCodeClassLoaderCreated = true; + } else { + if (!(runtimeEnv == null || runtimeEnv == RuntimeEnv.THREAD)) { + throw new IllegalStateException("The jar property must be specified in FunctionConfig."); + } + FunctionConfigUtils.validateJavaFunction(functionConfig, Thread.currentThread() + .getContextClassLoader()); } - } else if (functionConfig.getRuntime() == FunctionConfig.Runtime.GO) { userCodeFile = functionConfig.getGo(); } else if (functionConfig.getRuntime() == FunctionConfig.Runtime.PYTHON) { @@ -261,90 +288,89 @@ public void start(boolean blocking) throws Exception { throw new UnsupportedOperationException(); } - functionDetails = FunctionConfigUtils.convert(functionConfig, classLoader); + functionDetails = FunctionConfigUtils.convert(functionConfig, + userCodeClassLoader != null ? userCodeClassLoader : + Thread.currentThread().getContextClassLoader()); } else if (sourceConfig != null) { inferMissingArguments(sourceConfig); userCodeFile = sourceConfig.getArchive(); - - // if code file not specified try to get location of the code based on class. - if (userCodeFile == null && sourceConfig.getClassName() != null) { - userCodeFile = Thread.currentThread().getContextClassLoader() - .loadClass(sourceConfig.getClassName()) - .getProtectionDomain().getCodeSource().getLocation().getFile(); - } - - if (userCodeFile == null) { - userCodeFile = Thread.currentThread().getContextClassLoader() - .loadClass(LocalRunner.class.getName()) - .getProtectionDomain().getCodeSource().getLocation().getFile(); - } - parallelism = sourceConfig.getParallelism(); - ClassLoader builtInSourceClassLoader = isBuiltInSource(userCodeFile); + ClassLoader builtInSourceClassLoader = userCodeFile != null ? isBuiltInSource(userCodeFile) : null; if (builtInSourceClassLoader != null) { functionDetails = SourceConfigUtils.convert( sourceConfig, SourceConfigUtils.validateAndExtractDetails( sourceConfig, builtInSourceClassLoader, true)); - } else if (org.apache.pulsar.common.functions.Utils.isFunctionPackageUrlSupported(userCodeFile)) { + userCodeClassLoader = builtInSourceClassLoader; + } else if (userCodeFile != null && Utils.isFunctionPackageUrlSupported(userCodeFile)) { File file = FunctionCommon.extractFileFromPkgURL(userCodeFile); - ClassLoader classLoader = FunctionCommon.getClassLoaderFromPackage( + ClassLoader sourceClassLoader = FunctionCommon.getClassLoaderFromPackage( Function.FunctionDetails.ComponentType.SOURCE, sourceConfig.getClassName(), file, narExtractionDirectory); functionDetails = SourceConfigUtils.convert( - sourceConfig, SourceConfigUtils.validateAndExtractDetails(sourceConfig, classLoader, true)); - } else { + sourceConfig, + SourceConfigUtils.validateAndExtractDetails(sourceConfig, sourceClassLoader, true)); + userCodeClassLoader = sourceClassLoader; + userCodeClassLoaderCreated = true; + } else if (userCodeFile != null) { File file = new File(userCodeFile); if (!file.exists()) { throw new RuntimeException("Source archive (" + userCodeFile + ") does not exist"); } - ClassLoader classLoader = FunctionCommon.getClassLoaderFromPackage( + ClassLoader sourceClassLoader = FunctionCommon.getClassLoaderFromPackage( Function.FunctionDetails.ComponentType.SOURCE, sourceConfig.getClassName(), file, narExtractionDirectory); functionDetails = SourceConfigUtils.convert( - sourceConfig, SourceConfigUtils.validateAndExtractDetails(sourceConfig, classLoader, true)); + sourceConfig, SourceConfigUtils.validateAndExtractDetails(sourceConfig, sourceClassLoader, true)); + userCodeClassLoader = sourceClassLoader; + userCodeClassLoaderCreated = true; + } else { + if (!(runtimeEnv == null || runtimeEnv == RuntimeEnv.THREAD)) { + throw new IllegalStateException("The archive property must be specified in SourceConfig."); + } + functionDetails = SourceConfigUtils.convert( + sourceConfig, SourceConfigUtils.validateAndExtractDetails( + sourceConfig, Thread.currentThread().getContextClassLoader(), true)); } } else if (sinkConfig != null) { inferMissingArguments(sinkConfig); userCodeFile = sinkConfig.getArchive(); - - // if code file not specified try to get location of the code based on class. - if (userCodeFile == null && sinkConfig.getClassName() != null) { - userCodeFile = Thread.currentThread().getContextClassLoader() - .loadClass(sinkConfig.getClassName()) - .getProtectionDomain().getCodeSource().getLocation().getFile(); - } - - if (userCodeFile == null) { - userCodeFile = Thread.currentThread().getContextClassLoader() - .loadClass(LocalRunner.class.getName()) - .getProtectionDomain().getCodeSource().getLocation().getFile(); - } - parallelism = sinkConfig.getParallelism(); - ClassLoader builtInSinkClassLoader = isBuiltInSink(userCodeFile); + ClassLoader builtInSinkClassLoader = userCodeFile != null ? isBuiltInSink(userCodeFile) : null; if (builtInSinkClassLoader != null) { functionDetails = SinkConfigUtils.convert( sinkConfig, SinkConfigUtils.validateAndExtractDetails( sinkConfig, builtInSinkClassLoader, true)); - } else if (org.apache.pulsar.common.functions.Utils.isFunctionPackageUrlSupported(userCodeFile)) { + userCodeClassLoader = builtInSinkClassLoader; + } else if (Utils.isFunctionPackageUrlSupported(userCodeFile)) { File file = FunctionCommon.extractFileFromPkgURL(userCodeFile); - ClassLoader classLoader = FunctionCommon.getClassLoaderFromPackage( + ClassLoader sinkClassLoader = FunctionCommon.getClassLoaderFromPackage( Function.FunctionDetails.ComponentType.SINK, sinkConfig.getClassName(), file, narExtractionDirectory); functionDetails = SinkConfigUtils.convert( - sinkConfig, SinkConfigUtils.validateAndExtractDetails(sinkConfig, classLoader, true)); - } else { + sinkConfig, SinkConfigUtils.validateAndExtractDetails(sinkConfig, sinkClassLoader, true)); + userCodeClassLoader = sinkClassLoader; + userCodeClassLoaderCreated = true; + } else if (userCodeFile != null) { File file = new File(userCodeFile); if (!file.exists()) { throw new RuntimeException("Sink archive does not exist"); } - ClassLoader classLoader = FunctionCommon.getClassLoaderFromPackage( + ClassLoader sinkClassLoader = FunctionCommon.getClassLoaderFromPackage( Function.FunctionDetails.ComponentType.SINK, sinkConfig.getClassName(), file, narExtractionDirectory); functionDetails = SinkConfigUtils.convert( - sinkConfig, SinkConfigUtils.validateAndExtractDetails(sinkConfig, classLoader, true)); + sinkConfig, SinkConfigUtils.validateAndExtractDetails(sinkConfig, sinkClassLoader, true)); + userCodeClassLoader = sinkClassLoader; + userCodeClassLoaderCreated = true; + } else { + if (!(runtimeEnv == null || runtimeEnv == RuntimeEnv.THREAD)) { + throw new IllegalStateException("The archive property must be specified in SourceConfig."); + } + functionDetails = SinkConfigUtils.convert( + sinkConfig, SinkConfigUtils.validateAndExtractDetails( + sinkConfig, Thread.currentThread().getContextClassLoader(), true)); } } else { throw new IllegalArgumentException("Must specify Function, Source or Sink config"); @@ -484,13 +510,23 @@ private void startThreadedMode(org.apache.pulsar.functions.proto.Function.Functi if (functionConfig != null && functionConfig.getExposePulsarAdminClientEnabled() != null) { exposePulsarAdminClientEnabled = functionConfig.getExposePulsarAdminClientEnabled(); } - ThreadRuntimeFactory threadRuntimeFactory = new ThreadRuntimeFactory("LocalRunnerThreadGroup", - serviceUrl, - stateStorageServiceUrl, - authConfig, - secretsProvider, - null, narExtractionDirectory, null, - exposePulsarAdminClientEnabled, webServiceUrl); + ThreadRuntimeFactory threadRuntimeFactory; + ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader(); + try { + if (userCodeClassLoader != null) { + Thread.currentThread().setContextClassLoader(userCodeClassLoader); + } + threadRuntimeFactory = new ThreadRuntimeFactory("LocalRunnerThreadGroup", + serviceUrl, + stateStorageServiceUrl, + authConfig, + secretsProvider, + null, narExtractionDirectory, + null, + exposePulsarAdminClientEnabled, webServiceUrl); + } finally { + Thread.currentThread().setContextClassLoader(originalClassLoader); + } for (int i = 0; i < parallelism; ++i) { InstanceConfig instanceConfig = new InstanceConfig(); instanceConfig.setFunctionDetails(functionDetails); diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java index db569b123ef0e..ec9343e3d131d 100644 --- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java +++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java @@ -102,7 +102,6 @@ private static ClassLoader getFunctionClassLoader(InstanceConfig instanceConfig, String narExtractionDirectory, FunctionCacheManager fnCache, Optional connectorsManager) throws Exception { - if (FunctionCommon.isFunctionCodeBuiltin(instanceConfig.getFunctionDetails()) && connectorsManager.isPresent()) { switch (InstanceUtils.calculateSubjectType(instanceConfig.getFunctionDetails())) { @@ -124,6 +123,9 @@ private static ClassLoader loadJars(String jarFile, InstanceConfig instanceConfig, String narExtractionDirectory, FunctionCacheManager fnCache) throws Exception { + if (jarFile == null) { + return Thread.currentThread().getContextClassLoader(); + } ClassLoader fnClassLoader; try { log.info("Load JAR: {}", jarFile); @@ -175,6 +177,13 @@ public void start() throws Exception { String.format("%s-%s", FunctionCommon.getFullyQualifiedName(instanceConfig.getFunctionDetails()), instanceConfig.getInstanceId())); + this.fnThread.setContextClassLoader(functionClassLoader); + this.fnThread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { + @Override + public void uncaughtException(Thread t, Throwable e) { + log.error("Uncaught exception in thread {}", t, e); + } + }); this.fnThread.start(); } diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java index 35e43929b96c8..1be328bda6f70 100644 --- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java +++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java @@ -193,8 +193,9 @@ public void close() { } catch (PulsarClientException e) { log.warn("Failed to close pulsar client when closing function container factory", e); } - - pulsarAdmin.close(); + if (pulsarAdmin != null) { + pulsarAdmin.close(); + } // Shutdown instance cache InstanceCache.shutdown(); diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java index 6b9b7628f9332..c6a3cada08520 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java @@ -808,6 +808,11 @@ public static ClassLoader validate(FunctionConfig functionConfig, File functionP } } + public static void validateJavaFunction(FunctionConfig functionConfig, ClassLoader classLoader) { + doCommonChecks(functionConfig); + doJavaChecks(functionConfig, classLoader); + } + public static FunctionConfig validateUpdate(FunctionConfig existingConfig, FunctionConfig newConfig) { FunctionConfig mergedConfig = existingConfig.toBuilder().build(); if (!existingConfig.getTenant().equals(newConfig.getTenant())) { diff --git a/pulsar-functions/worker/pom.xml b/pulsar-functions/worker/pom.xml index c99544fd9ec40..eecacdfe2c337 100644 --- a/pulsar-functions/worker/pom.xml +++ b/pulsar-functions/worker/pom.xml @@ -140,6 +140,7 @@ ${project.groupId} pulsar-io-cassandra ${project.version} + pom test @@ -147,6 +148,15 @@ ${project.groupId} pulsar-io-twitter ${project.version} + pom + test + + + + ${project.groupId} + pulsar-functions-api-examples + ${project.version} + pom test @@ -154,7 +164,75 @@ - + + maven-dependency-plugin + + + copy-pulsar-io-connectors + generate-test-resources + + copy + + + + + ${project.groupId} + pulsar-io-cassandra + ${project.version} + jar + true + ${project.build.directory} + pulsar-io-cassandra.nar + + + ${project.groupId} + pulsar-io-twitter + ${project.version} + jar + true + ${project.build.directory} + pulsar-io-twitter.nar + + + ${project.groupId} + pulsar-functions-api-examples + ${project.version} + jar + true + ${project.build.directory} + pulsar-functions-api-examples.jar + + + + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + ${project.build.directory}/pulsar-io-data-generator.nar + ${project.build.directory}/pulsar-functions-api-examples.jar + + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + ${project.build.directory}/pulsar-io-cassandra.nar + ${project.build.directory}/pulsar-io-twitter.nar + + ${project.build.directory}/pulsar-functions-api-examples.jar + + + + org.apache.maven.plugins maven-antrun-plugin @@ -166,14 +244,6 @@ - copy test sink package - - - copy test source package - - copy test config files diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java index 6cf90606d4848..5f4665a296264 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java @@ -143,7 +143,7 @@ public void startFunction(FunctionRuntimeInfo functionRuntimeInfo) { } catch (Exception ex) { FunctionDetails details = functionRuntimeInfo.getFunctionInstance() .getFunctionMetaData().getFunctionDetails(); - log.info("{}/{}/{} Error starting function", details.getTenant(), details.getNamespace(), + log.error("{}/{}/{} Error starting function", details.getTenant(), details.getNamespace(), details.getName(), ex); functionRuntimeInfo.setStartupException(ex); } 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 b933c0d440160..db62afdb276d6 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 @@ -164,7 +164,7 @@ public void registerFunction(final String tenant, componentPackageFile = FunctionCommon.extractFileFromPkgURL(functionPkgUrl); } catch (Exception e) { - throw new IllegalArgumentException(String.format("Encountered error \"%s\" when getting %s package from %s", e.getMessage(), ComponentTypeUtils.toString(componentType), functionPkgUrl)); + throw new IllegalArgumentException(String.format("Encountered error \"%s\" when getting %s package from %s", e.getMessage(), ComponentTypeUtils.toString(componentType), functionPkgUrl), e); } } functionDetails = validateUpdateRequestParams(tenant, namespace, functionName, diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java index 74ebcd8a09039..473b25e2c2264 100644 --- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java +++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java @@ -18,7 +18,35 @@ */ package org.apache.pulsar.functions.worker.rest.api.v3; +import static org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.ATLEAST_ONCE; +import static org.apache.pulsar.functions.source.TopicSchema.DEFAULT_SERDE; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.when; +import static org.powermock.api.mockito.PowerMockito.doNothing; +import static org.powermock.api.mockito.PowerMockito.doReturn; +import static org.powermock.api.mockito.PowerMockito.doThrow; +import static org.powermock.api.mockito.PowerMockito.mockStatic; +import static org.testng.Assert.assertEquals; import com.google.common.collect.Lists; +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import javax.ws.rs.core.Response; import org.apache.distributedlog.api.namespace.Namespace; import org.apache.logging.log4j.Level; import org.apache.logging.log4j.core.config.Configurator; @@ -51,7 +79,6 @@ import org.apache.pulsar.functions.worker.WorkerConfig; import org.apache.pulsar.functions.worker.WorkerUtils; import org.apache.pulsar.functions.worker.rest.api.SinksImpl; -import org.apache.pulsar.io.cassandra.CassandraStringSink; import org.glassfish.jersey.media.multipart.FormDataContentDisposition; import org.mockito.Mockito; import org.powermock.api.mockito.PowerMockito; @@ -62,37 +89,13 @@ import org.testng.annotations.ObjectFactory; import org.testng.annotations.Test; -import javax.ws.rs.core.Response; -import java.io.File; -import java.io.FileInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.net.URL; -import java.util.HashMap; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; - -import static org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.ATLEAST_ONCE; -import static org.apache.pulsar.functions.source.TopicSchema.DEFAULT_SERDE; -import static org.mockito.Mockito.any; -import static org.mockito.Mockito.anyString; -import static org.mockito.Mockito.eq; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.spy; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.when; -import static org.powermock.api.mockito.PowerMockito.doNothing; -import static org.powermock.api.mockito.PowerMockito.doReturn; -import static org.powermock.api.mockito.PowerMockito.doThrow; -import static org.powermock.api.mockito.PowerMockito.mockStatic; -import static org.testng.Assert.assertEquals; - /** * Unit test of {@link SinksApiV3Resource}. */ -@PrepareForTest({WorkerUtils.class, SinkConfigUtils.class, ConnectorUtils.class, FunctionCommon.class, ClassLoaderUtils.class, InstanceUtils.class}) -@PowerMockIgnore({ "javax.management.*", "javax.ws.*", "org.apache.logging.log4j.*", "org.apache.pulsar.io.*", "java.io.*" }) +@PrepareForTest({WorkerUtils.class, SinkConfigUtils.class, ConnectorUtils.class, FunctionCommon.class, + ClassLoaderUtils.class, InstanceUtils.class}) +@PowerMockIgnore({"javax.management.*", "javax.ws.*", "org.apache.logging.log4j.*", "org.apache.pulsar.io.*", + "java.io.*"}) public class SinkApiV3ResourceTest { @@ -105,16 +108,14 @@ public IObjectFactory getObjectFactory() { private static final String namespace = "test-namespace"; private static final String sink = "test-sink"; private static final Map topicsToSerDeClassName = new HashMap<>(); + static { topicsToSerDeClassName.put("persistent://sample/standalone/ns1/test_src", DEFAULT_SERDE); } + private static final String subscriptionName = "test-subscription"; - private static final String className = CassandraStringSink.class.getName(); + private static final String CASSANDRA_STRING_SINK = "org.apache.pulsar.io.cassandra.CassandraStringSink"; private static final int parallelism = 1; - private static final String JAR_FILE_NAME = "pulsar-io-cassandra.nar"; - private static final String INVALID_JAR_FILE_NAME = "pulsar-io-twitter.nar"; - private String JAR_FILE_PATH; - private String INVALID_JAR_FILE_PATH; private PulsarWorkerService mockedWorkerService; private PulsarAdmin mockedPulsarAdmin; @@ -134,6 +135,30 @@ public IObjectFactory getObjectFactory() { private LeaderService mockedLeaderService; private Packages mockedPackages; + private static final String SYSTEM_PROPERTY_NAME_CASSANDRA_NAR_FILE_PATH = "pulsar-io-cassandra.nar.path"; + + public static File getPulsarIOCassandraNar() { + return new File(Objects.requireNonNull(System.getProperty(SYSTEM_PROPERTY_NAME_CASSANDRA_NAR_FILE_PATH) + , "pulsar-io-cassandra.nar file location must be specified with " + + SYSTEM_PROPERTY_NAME_CASSANDRA_NAR_FILE_PATH + " system property")); + } + + private static final String SYSTEM_PROPERTY_NAME_TWITTER_NAR_FILE_PATH = "pulsar-io-twitter.nar.path"; + + public static File getPulsarIOTwitterNar() { + return new File(Objects.requireNonNull(System.getProperty(SYSTEM_PROPERTY_NAME_TWITTER_NAR_FILE_PATH) + , "pulsar-io-twitter.nar file location must be specified with " + + SYSTEM_PROPERTY_NAME_TWITTER_NAR_FILE_PATH + " system property")); + } + + private static final String SYSTEM_PROPERTY_NAME_INVALID_NAR_FILE_PATH = "pulsar-io-invalid.nar.path"; + + public static File getPulsarIOInvalidNar() { + return new File(Objects.requireNonNull(System.getProperty(SYSTEM_PROPERTY_NAME_INVALID_NAR_FILE_PATH) + , "invalid nar file location must be specified with " + + SYSTEM_PROPERTY_NAME_INVALID_NAR_FILE_PATH + " system property")); + } + @BeforeMethod public void setup() throws Exception { this.mockedManager = mock(FunctionMetaDataManager.class); @@ -168,23 +193,20 @@ public void setup() throws Exception { when(mockedTenants.getTenantInfo(any())).thenReturn(mockedTenantInfo); when(mockedNamespaces.getNamespaces(any())).thenReturn(namespaceList); when(mockedLeaderService.isLeader()).thenReturn(true); - doNothing().when(mockedPackages).download(anyString(), anyString()); - - URL file = Thread.currentThread().getContextClassLoader().getResource(JAR_FILE_NAME); - if (file == null) { - throw new RuntimeException("Failed to file required test archive: " + JAR_FILE_NAME); - } - JAR_FILE_PATH = file.getFile(); - INVALID_JAR_FILE_PATH = Thread.currentThread().getContextClassLoader().getResource(INVALID_JAR_FILE_NAME).getFile(); + doAnswer(invocationOnMock -> { + Files.copy(getPulsarIOCassandraNar().toPath(), Paths.get(invocationOnMock.getArgument(1, String.class)), + StandardCopyOption.REPLACE_EXISTING); + return null; + }).when(mockedPackages).download(any(), any()); // worker config WorkerConfig workerConfig = new WorkerConfig() - .setWorkerId("test") - .setWorkerPort(8080) - .setDownloadDirectory("/tmp/pulsar/functions") - .setFunctionMetadataTopicName("pulsar/functions") - .setNumFunctionPackageReplicas(3) - .setPulsarServiceUrl("pulsar://localhost:6650/"); + .setWorkerId("test") + .setWorkerPort(8080) + .setDownloadDirectory("/tmp/pulsar/functions") + .setFunctionMetadataTopicName("pulsar/functions") + .setNumFunctionPackageReplicas(3) + .setPulsarServiceUrl("pulsar://localhost:6650/"); when(mockedWorkerService.getWorkerConfig()).thenReturn(workerConfig); this.resource = spy(new SinksImpl(() -> mockedWorkerService)); @@ -200,17 +222,17 @@ public void setup() throws Exception { public void testRegisterSinkMissingTenant() { try { testRegisterSinkMissingArguments( - null, - namespace, + null, + namespace, sink, - mockedInputStream, - mockedFormData, - topicsToSerDeClassName, - className, - parallelism, + mockedInputStream, + mockedFormData, + topicsToSerDeClassName, + CASSANDRA_STRING_SINK, + parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -220,17 +242,17 @@ public void testRegisterSinkMissingTenant() { public void testRegisterSinkMissingNamespace() { try { testRegisterSinkMissingArguments( - tenant, - null, + tenant, + null, sink, - mockedInputStream, - mockedFormData, - topicsToSerDeClassName, - className, - parallelism, + mockedInputStream, + mockedFormData, + topicsToSerDeClassName, + CASSANDRA_STRING_SINK, + parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -240,16 +262,16 @@ public void testRegisterSinkMissingNamespace() { public void testRegisterSinkMissingSinkName() { try { testRegisterSinkMissingArguments( - tenant, - namespace, - null, - mockedInputStream, - mockedFormData, - topicsToSerDeClassName, - className, - parallelism, + tenant, + namespace, + null, + mockedInputStream, + mockedFormData, + topicsToSerDeClassName, + CASSANDRA_STRING_SINK, + parallelism, null); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -259,160 +281,174 @@ public void testRegisterSinkMissingSinkName() { public void testRegisterSinkMissingPackage() { try { testRegisterSinkMissingArguments( - tenant, - namespace, - sink, - null, - mockedFormData, - topicsToSerDeClassName, - null, - parallelism, - null + tenant, + namespace, + sink, + null, + mockedFormData, + topicsToSerDeClassName, + null, + parallelism, + null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink class UnknownClass must be in class path") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink class UnknownClass must " + + "be in class path") public void testRegisterSinkWrongClassName() { - try { - testRegisterSinkMissingArguments( - tenant, - namespace, - sink, - mockedInputStream, - mockedFormData, - topicsToSerDeClassName, - "UnknownClass", - parallelism, - null - ); - } catch (RestException re){ + try { + testRegisterSinkMissingArguments( + tenant, + namespace, + sink, + mockedInputStream, + mockedFormData, + topicsToSerDeClassName, + "UnknownClass", + parallelism, + null + ); + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink package does not have the" + - " correct format. Pulsar cannot determine if the package is a NAR package" + - " or JAR package. Sink classname is not provided and attempts to load it as a NAR package produced the following error.") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink package does not have the" + + " correct format. Pulsar cannot determine if the package is a NAR package" + + " or JAR package. Sink classname is not provided and attempts to load it as a NAR package produced the " + + "following error.") public void testRegisterSinkMissingPackageDetails() { try { testRegisterSinkMissingArguments( - tenant, - namespace, - sink, - mockedInputStream, - null, - topicsToSerDeClassName, - null, - parallelism, - null + tenant, + namespace, + sink, + mockedInputStream, + null, + topicsToSerDeClassName, + null, + parallelism, + null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Failed to extract sink class from archive") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Failed to extract sink class " + + "from archive") public void testRegisterSinkInvalidJarNoSink() throws IOException { try { - FileInputStream inputStream = new FileInputStream(INVALID_JAR_FILE_PATH); - testRegisterSinkMissingArguments( - tenant, - namespace, - sink, - inputStream, - mockedFormData, - topicsToSerDeClassName, - null, - parallelism, - null - ); - } catch (RestException re){ + try (FileInputStream inputStream = new FileInputStream(getPulsarIOTwitterNar())) { + testRegisterSinkMissingArguments( + tenant, + namespace, + sink, + inputStream, + mockedFormData, + topicsToSerDeClassName, + null, + parallelism, + null + ); + } + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Must specify at least one topic of input via topicToSerdeClassName, topicsPattern, topicToSchemaType or inputSpecs") - public void testRegisterSinkNoInput() { + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Must specify at least one " + + "topic of input via topicToSerdeClassName, topicsPattern, topicToSchemaType or inputSpecs") + public void testRegisterSinkNoInput() throws IOException { try { - testRegisterSinkMissingArguments( - tenant, - namespace, - sink, - mockedInputStream, - mockedFormData, - null, - className, - parallelism, - null - ); - } catch (RestException re){ + try (FileInputStream inputStream = new FileInputStream(getPulsarIOCassandraNar())) { + testRegisterSinkMissingArguments( + tenant, + namespace, + sink, + inputStream, + mockedFormData, + null, + CASSANDRA_STRING_SINK, + parallelism, + null + ); + } + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink parallelism must be a positive number") - public void testRegisterSinkNegativeParallelism() { + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink parallelism must be a " + + "positive number") + public void testRegisterSinkNegativeParallelism() throws IOException { try { - testRegisterSinkMissingArguments( - tenant, - namespace, - sink, - mockedInputStream, - mockedFormData, - topicsToSerDeClassName, - className, - -2, - null - ); - } catch (RestException re){ + try (FileInputStream inputStream = new FileInputStream(getPulsarIOCassandraNar())) { + testRegisterSinkMissingArguments( + tenant, + namespace, + sink, + inputStream, + mockedFormData, + topicsToSerDeClassName, + CASSANDRA_STRING_SINK, + -2, + null + ); + } + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink parallelism must be a positive number") - public void testRegisterSinkZeroParallelism() { + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink parallelism must be a " + + "positive number") + public void testRegisterSinkZeroParallelism() throws IOException { try { - testRegisterSinkMissingArguments( - tenant, - namespace, - sink, - mockedInputStream, - mockedFormData, - topicsToSerDeClassName, - className, - 0, - null - ); - } catch (RestException re){ + try (FileInputStream inputStream = new FileInputStream(getPulsarIOCassandraNar())) { + testRegisterSinkMissingArguments( + tenant, + namespace, + sink, + inputStream, + mockedFormData, + topicsToSerDeClassName, + CASSANDRA_STRING_SINK, + 0, + null + ); + } + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Encountered error .*. when getting Sink package from .*") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Encountered error .*. when " + + "getting Sink package from .*") public void testRegisterSinkHttpUrl() { try { testRegisterSinkMissingArguments( - tenant, - namespace, - sink, - null, - null, - topicsToSerDeClassName, - className, - parallelism, - "http://localhost:1234/test" + tenant, + namespace, + sink, + null, + null, + topicsToSerDeClassName, + CASSANDRA_STRING_SINK, + parallelism, + "http://localhost:1234/test" ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -493,15 +529,17 @@ private void registerDefaultSink() throws IOException { private void registerDefaultSinkWithPackageUrl(String packageUrl) throws IOException { SinkConfig sinkConfig = createDefaultSinkConfig(); - resource.registerSink( - tenant, - namespace, - sink, - new FileInputStream(JAR_FILE_PATH), - mockedFormData, - packageUrl, - sinkConfig, - null, null); + try (FileInputStream inputStream = new FileInputStream(getPulsarIOCassandraNar())) { + resource.registerSink( + tenant, + namespace, + sink, + inputStream, + mockedFormData, + packageUrl, + sinkConfig, + null, null); + } } @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink test-sink already exists") @@ -512,7 +550,7 @@ public void testRegisterExistedSink() throws IOException { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(sink))).thenReturn(true); registerDefaultSink(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -525,15 +563,15 @@ public void testRegisterSinkUploadFailure() throws Exception { doThrow(new IOException("upload failure")).when(WorkerUtils.class); WorkerUtils.uploadFileToBookkeeper( anyString(), - any(File.class), - any(Namespace.class)); + any(File.class), + any(Namespace.class)); PowerMockito.when(WorkerUtils.class, "dumpToTmpFile", any()).thenCallRealMethod(); when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(sink))).thenReturn(false); registerDefaultSink(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.INTERNAL_SERVER_ERROR); throw re; } @@ -578,18 +616,20 @@ public void testRegisterSinkConflictingFields() throws Exception { sinkConfig.setTenant(tenant); sinkConfig.setNamespace(namespace); sinkConfig.setName(sink); - sinkConfig.setClassName(className); + sinkConfig.setClassName(CASSANDRA_STRING_SINK); sinkConfig.setParallelism(parallelism); sinkConfig.setTopicToSerdeClassName(topicsToSerDeClassName); - resource.registerSink( - actualTenant, - actualNamespace, - actualName, - new FileInputStream(JAR_FILE_PATH), - mockedFormData, - null, - sinkConfig, - null, null); + try (FileInputStream inputStream = new FileInputStream(getPulsarIOCassandraNar())) { + resource.registerSink( + actualTenant, + actualNamespace, + actualName, + inputStream, + mockedFormData, + null, + sinkConfig, + null, null); + } } @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "sink failed to register") @@ -610,13 +650,14 @@ public void testRegisterSinkFailure() throws Exception { .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); registerDefaultSink(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function registration interrupted") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function registration " + + "interrupted") public void testRegisterSinkInterrupted() throws Exception { try { mockStatic(WorkerUtils.class); @@ -634,7 +675,7 @@ public void testRegisterSinkInterrupted() throws Exception { .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); registerDefaultSink(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.INTERNAL_SERVER_ERROR); throw re; } @@ -649,7 +690,7 @@ public void testRegisterSinkSuccessWithPackageName() throws IOException { public void testRegisterSinkFailedWithWrongPackageName() throws PulsarAdminException, IOException { try { doThrow(new PulsarAdminException("package name is invalid")) - .when(mockedPackages).download(anyString(), anyString()); + .when(mockedPackages).download(anyString(), anyString()); registerDefaultSinkWithPackageUrl("function://"); } catch (RestException e) { // expected exception @@ -665,16 +706,16 @@ public void testRegisterSinkFailedWithWrongPackageName() throws PulsarAdminExcep public void testUpdateSinkMissingTenant() throws Exception { try { testUpdateSinkMissingArguments( - null, - namespace, - sink, - mockedInputStream, - mockedFormData, - topicsToSerDeClassName, - className, - parallelism, - "Tenant is not provided"); - } catch (RestException re){ + null, + namespace, + sink, + mockedInputStream, + mockedFormData, + topicsToSerDeClassName, + CASSANDRA_STRING_SINK, + parallelism, + "Tenant is not provided"); + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -684,16 +725,16 @@ public void testUpdateSinkMissingTenant() throws Exception { public void testUpdateSinkMissingNamespace() throws Exception { try { testUpdateSinkMissingArguments( - tenant, - null, - sink, - mockedInputStream, - mockedFormData, - topicsToSerDeClassName, - className, - parallelism, - "Namespace is not provided"); - } catch (RestException re){ + tenant, + null, + sink, + mockedInputStream, + mockedFormData, + topicsToSerDeClassName, + CASSANDRA_STRING_SINK, + parallelism, + "Namespace is not provided"); + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -703,16 +744,16 @@ public void testUpdateSinkMissingNamespace() throws Exception { public void testUpdateSinkMissingFunctionName() throws Exception { try { testUpdateSinkMissingArguments( - tenant, - namespace, - null, - mockedInputStream, - mockedFormData, - topicsToSerDeClassName, - className, - parallelism, - "Sink name is not provided"); - } catch (RestException re){ + tenant, + namespace, + null, + mockedInputStream, + mockedFormData, + topicsToSerDeClassName, + CASSANDRA_STRING_SINK, + parallelism, + "Sink name is not provided"); + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -728,16 +769,16 @@ public void testUpdateSinkMissingPackage() throws Exception { PowerMockito.when(WorkerUtils.class, "dumpToTmpFile", any()).thenCallRealMethod(); testUpdateSinkMissingArguments( - tenant, - namespace, + tenant, + namespace, sink, - null, - mockedFormData, - topicsToSerDeClassName, - null, - parallelism, + null, + mockedFormData, + topicsToSerDeClassName, + null, + parallelism, "Update contains no change"); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -753,16 +794,16 @@ public void testUpdateSinkMissingInputs() throws Exception { PowerMockito.when(WorkerUtils.class, "dumpToTmpFile", any()).thenCallRealMethod(); testUpdateSinkMissingArguments( - tenant, - namespace, - sink, - null, - mockedFormData, - null, - null, - parallelism, - "Update contains no change"); - } catch (RestException re){ + tenant, + namespace, + sink, + null, + mockedFormData, + null, + null, + parallelism, + "Update contains no change"); + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -780,16 +821,16 @@ public void testUpdateSinkDifferentInputs() throws Exception { Map inputTopics = new HashMap<>(); inputTopics.put("DifferntTopic", DEFAULT_SERDE); testUpdateSinkMissingArguments( - tenant, - namespace, - sink, - null, - mockedFormData, - inputTopics, - className, - parallelism, - "Input Topics cannot be altered"); - } catch (RestException re){ + tenant, + namespace, + sink, + null, + mockedFormData, + inputTopics, + CASSANDRA_STRING_SINK, + parallelism, + "Input Topics cannot be altered"); + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -810,7 +851,7 @@ public void testUpdateSinkDifferentParallelism() throws Exception { null, mockedFormData, topicsToSerDeClassName, - className, + CASSANDRA_STRING_SINK, parallelism + 1, null); } @@ -826,14 +867,15 @@ private void testUpdateSinkMissingArguments( Integer parallelism, String expectedError) throws Exception { mockStatic(ConnectorUtils.class); - doReturn(CassandraStringSink.class.getName()).when(ConnectorUtils.class); + doReturn(CASSANDRA_STRING_SINK).when(ConnectorUtils.class); ConnectorUtils.getIOSinkClass(any(NarClassLoader.class)); mockStatic(ClassLoaderUtils.class); mockStatic(FunctionCommon.class); PowerMockito.when(FunctionCommon.class, "createPkgTempFile").thenCallRealMethod(); - PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()).thenCallRealMethod(); + PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()) + .thenCallRealMethod(); doReturn(String.class).when(FunctionCommon.class); FunctionCommon.getSinkType(any()); @@ -844,7 +886,8 @@ private void testUpdateSinkMissingArguments( doReturn(ATLEAST_ONCE).when(FunctionCommon.class); FunctionCommon.convertProcessingGuarantee(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE); - this.mockedFunctionMetaData = FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); + this.mockedFunctionMetaData = + FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); when(mockedManager.getFunctionMetaData(any(), any(), any())).thenReturn(mockedFunctionMetaData); when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(sink))).thenReturn(true); @@ -875,13 +918,13 @@ private void testUpdateSinkMissingArguments( } resource.updateSink( - tenant, - namespace, - sink, - inputStream, - details, - null, - sinkConfig, + tenant, + namespace, + sink, + inputStream, + details, + null, + sinkConfig, null, null, null); } @@ -895,19 +938,20 @@ private void updateDefaultSinkWithPackageUrl(String packageUrl) throws Exception sinkConfig.setTenant(tenant); sinkConfig.setNamespace(namespace); sinkConfig.setName(sink); - sinkConfig.setClassName(className); + sinkConfig.setClassName(CASSANDRA_STRING_SINK); sinkConfig.setParallelism(parallelism); sinkConfig.setTopicToSerdeClassName(topicsToSerDeClassName); mockStatic(ConnectorUtils.class); - doReturn(CassandraStringSink.class.getName()).when(ConnectorUtils.class); + doReturn(CASSANDRA_STRING_SINK).when(ConnectorUtils.class); ConnectorUtils.getIOSinkClass(any(NarClassLoader.class)); mockStatic(ClassLoaderUtils.class); mockStatic(FunctionCommon.class); PowerMockito.when(FunctionCommon.class, "createPkgTempFile").thenCallRealMethod(); - PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()).thenCallRealMethod(); + PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()) + .thenCallRealMethod(); doReturn(String.class).when(FunctionCommon.class); FunctionCommon.getSinkType(any()); @@ -919,18 +963,21 @@ private void updateDefaultSinkWithPackageUrl(String packageUrl) throws Exception FunctionCommon.convertProcessingGuarantee(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE); - this.mockedFunctionMetaData = FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); + this.mockedFunctionMetaData = + FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); when(mockedManager.getFunctionMetaData(any(), any(), any())).thenReturn(mockedFunctionMetaData); - resource.updateSink( - tenant, - namespace, - sink, - new FileInputStream(JAR_FILE_PATH), - mockedFormData, - packageUrl, - sinkConfig, - null, null, null); + try (FileInputStream inputStream = new FileInputStream(getPulsarIOCassandraNar())) { + resource.updateSink( + tenant, + namespace, + sink, + inputStream, + mockedFormData, + packageUrl, + sinkConfig, + null, null, null); + } } @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink test-sink doesn't exist") @@ -984,19 +1031,19 @@ public void testUpdateSinkSuccess() throws Exception { public void testUpdateSinkWithUrl() throws Exception { Configurator.setRootLevel(Level.DEBUG); - String filePackageUrl = "file://" + JAR_FILE_PATH; + String filePackageUrl = getPulsarIOCassandraNar().toURI().toString(); SinkConfig sinkConfig = new SinkConfig(); sinkConfig.setTopicToSerdeClassName(topicsToSerDeClassName); sinkConfig.setTenant(tenant); sinkConfig.setNamespace(namespace); sinkConfig.setName(sink); - sinkConfig.setClassName(className); + sinkConfig.setClassName(CASSANDRA_STRING_SINK); sinkConfig.setParallelism(parallelism); when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(sink))).thenReturn(true); mockStatic(ConnectorUtils.class); - doReturn(CassandraStringSink.class.getName()).when(ConnectorUtils.class); + doReturn(CASSANDRA_STRING_SINK).when(ConnectorUtils.class); ConnectorUtils.getIOSinkClass(any(NarClassLoader.class)); mockStatic(ClassLoaderUtils.class); @@ -1005,7 +1052,8 @@ public void testUpdateSinkWithUrl() throws Exception { doReturn(String.class).when(FunctionCommon.class); FunctionCommon.getSinkType(any()); PowerMockito.when(FunctionCommon.class, "extractFileFromPkgURL", any()).thenCallRealMethod(); - PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()).thenCallRealMethod(); + PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()) + .thenCallRealMethod(); doReturn(mock(NarClassLoader.class)).when(FunctionCommon.class); FunctionCommon.extractNarClassLoader(any(), any()); @@ -1013,17 +1061,18 @@ public void testUpdateSinkWithUrl() throws Exception { doReturn(ATLEAST_ONCE).when(FunctionCommon.class); FunctionCommon.convertProcessingGuarantee(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE); - this.mockedFunctionMetaData = FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); + this.mockedFunctionMetaData = + FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); when(mockedManager.getFunctionMetaData(any(), any(), any())).thenReturn(mockedFunctionMetaData); resource.updateSink( - tenant, - namespace, + tenant, + namespace, sink, - null, - null, - filePackageUrl, - sinkConfig, + null, + null, + filePackageUrl, + sinkConfig, null, null, null); } @@ -1061,7 +1110,7 @@ public void testUpdateSinkFailedWithWrongPackageName() throws Exception { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(sink))).thenReturn(true); try { doThrow(new PulsarAdminException("package name is invalid")) - .when(mockedPackages).download(anyString(), anyString()); + .when(mockedPackages).download(anyString(), anyString()); updateDefaultSinkWithPackageUrl("function://"); } catch (RestException e) { // expected exception @@ -1069,7 +1118,8 @@ public void testUpdateSinkFailedWithWrongPackageName() throws Exception { } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function registration interrupted") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function registration " + + "interrupted") public void testUpdateSinkInterrupted() throws Exception { try { mockStatic(WorkerUtils.class); @@ -1100,9 +1150,9 @@ public void testUpdateSinkInterrupted() throws Exception { public void testDeregisterSinkMissingTenant() { try { testDeregisterSinkMissingArguments( - null, - namespace, - sink + null, + namespace, + sink ); } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); @@ -1114,8 +1164,8 @@ public void testDeregisterSinkMissingTenant() { public void testDeregisterSinkMissingNamespace() { try { testDeregisterSinkMissingArguments( - tenant, - null, + tenant, + null, sink ); } catch (RestException re) { @@ -1128,9 +1178,9 @@ public void testDeregisterSinkMissingNamespace() { public void testDeregisterSinkMissingFunctionName() { try { testDeregisterSinkMissingArguments( - tenant, - namespace, - null + tenant, + namespace, + null ); } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); @@ -1144,23 +1194,23 @@ private void testDeregisterSinkMissingArguments( String sink ) { resource.deregisterFunction( - tenant, - namespace, - sink, + tenant, + namespace, + sink, null, null); } private void deregisterDefaultSink() { resource.deregisterFunction( - tenant, - namespace, + tenant, + namespace, sink, null, null); } @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Sink test-sink doesn't exist") - public void testDeregisterNotExistedSink() { + public void testDeregisterNotExistedSink() { try { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(sink))).thenReturn(false); deregisterDefaultSink(); @@ -1180,7 +1230,8 @@ public void testDeregisterSinkFailure() throws Exception { try { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(sink))).thenReturn(true); - when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(sink))).thenReturn(FunctionMetaData.newBuilder().build()); + when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(sink))) + .thenReturn(FunctionMetaData.newBuilder().build()); doThrow(new IllegalArgumentException("sink failed to deregister")) .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); @@ -1192,12 +1243,14 @@ public void testDeregisterSinkFailure() throws Exception { } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function deregistration interrupted") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function deregistration " + + "interrupted") public void testDeregisterSinkInterrupted() throws Exception { try { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(sink))).thenReturn(true); - when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(sink))).thenReturn(FunctionMetaData.newBuilder().build()); + when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(sink))) + .thenReturn(FunctionMetaData.newBuilder().build()); doThrow(new IllegalStateException("Function deregistration interrupted")) .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); @@ -1216,7 +1269,8 @@ public void testDeregisterSinkBKPackageCleanup() throws IOException { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(sink))).thenReturn(true); - String packagePath = "public/default/test/591541f0-c7c5-40c0-983b-610c722f90b0-pulsar-io-batch-data-generator-2.7.0.nar"; + String packagePath = + "public/default/test/591541f0-c7c5-40c0-983b-610c722f90b0-pulsar-io-batch-data-generator-2.7.0.nar"; when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(sink))) .thenReturn(FunctionMetaData.newBuilder().setPackageLocation( Function.PackageLocationMetaData.newBuilder().setPackagePath(packagePath).build()).build()); @@ -1292,8 +1346,8 @@ public void testDeregisterFileSinkBKPackageCleanup() throws IOException { public void testGetSinkMissingTenant() { try { testGetSinkMissingArguments( - null, - namespace, + null, + namespace, sink ); } catch (RestException re) { @@ -1321,9 +1375,9 @@ public void testGetSinkMissingFunctionName() { try { testGetSinkMissingArguments( - tenant, - namespace, - null + tenant, + namespace, + null ); } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); @@ -1337,17 +1391,17 @@ private void testGetSinkMissingArguments( String sink ) { resource.getFunctionInfo( - tenant, - namespace, - sink, null, null + tenant, + namespace, + sink, null, null ); } private SinkConfig getDefaultSinkInfo() { return resource.getSinkInfo( - tenant, - namespace, + tenant, + namespace, sink ); } @@ -1371,9 +1425,9 @@ public void testGetSinkSuccess() { .setSubscriptionType(Function.SubscriptionType.SHARED) .setSubscriptionName(subscriptionName) .putInputSpecs("input", Function.ConsumerSpec.newBuilder() - .setSerdeClassName(DEFAULT_SERDE) - .setIsRegexPattern(false) - .build()).build(); + .setSerdeClassName(DEFAULT_SERDE) + .setIsRegexPattern(false) + .build()).build(); Function.SinkSpec sinkSpec = Function.SinkSpec.newBuilder() .setBuiltin("jdbc") .build(); @@ -1388,17 +1442,17 @@ public void testGetSinkSuccess() { .setRuntime(FunctionDetails.Runtime.JAVA) .setSource(sourceSpec).build(); FunctionMetaData metaData = FunctionMetaData.newBuilder() - .setCreateTime(System.currentTimeMillis()) - .setFunctionDetails(functionDetails) - .setPackageLocation(Function.PackageLocationMetaData.newBuilder().setPackagePath("/path/to/package")) - .setVersion(1234) - .build(); + .setCreateTime(System.currentTimeMillis()) + .setFunctionDetails(functionDetails) + .setPackageLocation(Function.PackageLocationMetaData.newBuilder().setPackagePath("/path/to/package")) + .setVersion(1234) + .build(); when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(sink))).thenReturn(metaData); getDefaultSinkInfo(); assertEquals( - SinkConfigUtils.convertFromDetails(functionDetails), - getDefaultSinkInfo()); + SinkConfigUtils.convertFromDetails(functionDetails), + getDefaultSinkInfo()); } // @@ -1409,10 +1463,10 @@ public void testGetSinkSuccess() { public void testListSinksMissingTenant() { try { testListSinksMissingArguments( - null, - namespace + null, + namespace ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1422,10 +1476,10 @@ public void testListSinksMissingTenant() { public void testListFunctionsMissingNamespace() { try { testListSinksMissingArguments( - tenant, - null + tenant, + null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1436,16 +1490,16 @@ private void testListSinksMissingArguments( String namespace ) { resource.listFunctions( - tenant, - namespace, null, null + tenant, + namespace, null, null ); } private List listDefaultSinks() { return resource.listFunctions( - tenant, - namespace, null, null + tenant, + namespace, null, null ); } @@ -1481,9 +1535,12 @@ public void testOnlyGetSinks() { when(mockedManager.listFunctions(eq(tenant), eq(namespace))).thenReturn(functionMetaDataList); mockStatic(InstanceUtils.class); - PowerMockito.when(InstanceUtils.calculateSubjectType(f1.getFunctionDetails())).thenReturn(FunctionDetails.ComponentType.SOURCE); - PowerMockito.when(InstanceUtils.calculateSubjectType(f2.getFunctionDetails())).thenReturn(FunctionDetails.ComponentType.FUNCTION); - PowerMockito.when(InstanceUtils.calculateSubjectType(f3.getFunctionDetails())).thenReturn(FunctionDetails.ComponentType.SINK); + PowerMockito.when(InstanceUtils.calculateSubjectType(f1.getFunctionDetails())) + .thenReturn(FunctionDetails.ComponentType.SOURCE); + PowerMockito.when(InstanceUtils.calculateSubjectType(f2.getFunctionDetails())) + .thenReturn(FunctionDetails.ComponentType.FUNCTION); + PowerMockito.when(InstanceUtils.calculateSubjectType(f3.getFunctionDetails())) + .thenReturn(FunctionDetails.ComponentType.SINK); List sinkList = listDefaultSinks(); assertEquals(functions, sinkList); @@ -1494,7 +1551,7 @@ public void testregisterSinkNonExistingNamespace() throws Exception { try { this.namespaceList.clear(); registerDefaultSink(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1505,7 +1562,7 @@ public void testregisterSinkNonExistingTenant() throws Exception { try { when(mockedTenants.getTenantInfo(any())).thenThrow(PulsarAdminException.NotFoundException.class); registerDefaultSink(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1516,7 +1573,7 @@ private SinkConfig createDefaultSinkConfig() { sinkConfig.setTenant(tenant); sinkConfig.setNamespace(namespace); sinkConfig.setName(sink); - sinkConfig.setClassName(className); + sinkConfig.setClassName(CASSANDRA_STRING_SINK); sinkConfig.setParallelism(parallelism); sinkConfig.setTopicToSerdeClassName(topicsToSerDeClassName); return sinkConfig; diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java index 4113dffe2f7bd..cd85ec6eb4059 100644 --- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java +++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java @@ -18,9 +18,14 @@ */ package org.apache.pulsar.functions.worker.rest.api.v3; +import static org.apache.pulsar.functions.worker.rest.api.v3.SinkApiV3ResourceTest.getPulsarIOCassandraNar; +import static org.apache.pulsar.functions.worker.rest.api.v3.SinkApiV3ResourceTest.getPulsarIOInvalidNar; +import static org.apache.pulsar.functions.worker.rest.api.v3.SinkApiV3ResourceTest.getPulsarIOTwitterNar; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; @@ -30,19 +35,18 @@ import static org.powermock.api.mockito.PowerMockito.doThrow; import static org.powermock.api.mockito.PowerMockito.mockStatic; import static org.testng.Assert.assertEquals; - import com.google.common.collect.Lists; - import java.io.File; import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; -import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; +import java.util.Collections; import java.util.LinkedList; import java.util.List; - import javax.ws.rs.core.Response; - import org.apache.distributedlog.api.namespace.Namespace; import org.apache.logging.log4j.Level; import org.apache.logging.log4j.core.config.Configurator; @@ -71,15 +75,21 @@ import org.apache.pulsar.functions.utils.FunctionCommon; import org.apache.pulsar.functions.utils.SourceConfigUtils; import org.apache.pulsar.functions.utils.io.ConnectorUtils; -import org.apache.pulsar.functions.worker.*; +import org.apache.pulsar.functions.worker.FunctionMetaDataManager; +import org.apache.pulsar.functions.worker.FunctionRuntimeManager; +import org.apache.pulsar.functions.worker.LeaderService; +import org.apache.pulsar.functions.worker.PulsarWorkerService; +import org.apache.pulsar.functions.worker.WorkerConfig; +import org.apache.pulsar.functions.worker.WorkerUtils; import org.apache.pulsar.functions.worker.rest.api.SourcesImpl; -import org.apache.pulsar.io.twitter.TwitterFireHose; import org.glassfish.jersey.media.multipart.FormDataContentDisposition; 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.IObjectFactory; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; import org.testng.annotations.BeforeMethod; import org.testng.annotations.ObjectFactory; import org.testng.annotations.Test; @@ -87,8 +97,9 @@ /** * Unit test of {@link SourcesApiV3Resource}. */ -@PrepareForTest({WorkerUtils.class, ConnectorUtils.class, FunctionCommon.class, ClassLoaderUtils.class, InstanceUtils.class}) -@PowerMockIgnore({ "javax.management.*", "javax.ws.*", "org.apache.logging.log4j.*", "org.apache.pulsar.io.*" }) +@PrepareForTest({WorkerUtils.class, ConnectorUtils.class, FunctionCommon.class, ClassLoaderUtils.class, + InstanceUtils.class}) +@PowerMockIgnore({"javax.management.*", "javax.ws.*", "org.apache.logging.log4j.*", "org.apache.pulsar.io.*"}) public class SourceApiV3ResourceTest { @ObjectFactory @@ -101,12 +112,8 @@ public IObjectFactory getObjectFactory() { private static final String source = "test-source"; private static final String outputTopic = "test-output-topic"; private static final String outputSerdeClassName = TopicSchema.DEFAULT_SERDE; - private static final String className = TwitterFireHose.class.getName(); + private static final String TWITTER_FIRE_HOSE = "org.apache.pulsar.io.twitter.TwitterFireHose"; private static final int parallelism = 1; - private static final String JAR_FILE_NAME = "pulsar-io-twitter.nar"; - private static final String INVALID_JAR_FILE_NAME = "pulsar-io-cassandra.nar"; - private String JAR_FILE_PATH; - private String INVALID_JAR_FILE_PATH; private PulsarWorkerService mockedWorkerService; private PulsarAdmin mockedPulsarAdmin; @@ -126,6 +133,21 @@ public IObjectFactory getObjectFactory() { private LeaderService mockedLeaderService; private Packages mockedPackages; + private static NarClassLoader narClassLoader; + + @BeforeClass + public void setupNarClassLoader() throws IOException { + narClassLoader = NarClassLoader.getFromArchive(getPulsarIOTwitterNar(), Collections.emptySet()); + } + + @AfterClass(alwaysRun = true) + public void cleanupNarClassLoader() throws IOException { + if (narClassLoader != null) { + narClassLoader.close(); + narClassLoader = null; + } + } + @BeforeMethod public void setup() throws Exception { this.mockedManager = mock(FunctionMetaDataManager.class); @@ -160,23 +182,20 @@ public void setup() throws Exception { when(mockedTenants.getTenantInfo(any())).thenReturn(mockedTenantInfo); when(mockedNamespaces.getNamespaces(any())).thenReturn(namespaceList); when(mockedLeaderService.isLeader()).thenReturn(true); - doNothing().when(mockedPackages).download(anyString(), anyString()); - - URL file = Thread.currentThread().getContextClassLoader().getResource(JAR_FILE_NAME); - if (file == null) { - throw new RuntimeException("Failed to file required test archive: " + JAR_FILE_NAME); - } - JAR_FILE_PATH = file.getFile(); - INVALID_JAR_FILE_PATH = Thread.currentThread().getContextClassLoader().getResource(INVALID_JAR_FILE_NAME).getFile(); + doAnswer(invocationOnMock -> { + Files.copy(getPulsarIOTwitterNar().toPath(), Paths.get(invocationOnMock.getArgument(1, String.class)), + StandardCopyOption.REPLACE_EXISTING); + return null; + }).when(mockedPackages).download(any(), any()); // worker config WorkerConfig workerConfig = new WorkerConfig() - .setWorkerId("test") - .setWorkerPort(8080) - .setDownloadDirectory("/tmp/pulsar/functions") - .setFunctionMetadataTopicName("pulsar/functions") - .setNumFunctionPackageReplicas(3) - .setPulsarServiceUrl("pulsar://localhost:6650/"); + .setWorkerId("test") + .setWorkerPort(8080) + .setDownloadDirectory("/tmp/pulsar/functions") + .setFunctionMetadataTopicName("pulsar/functions") + .setNumFunctionPackageReplicas(3) + .setPulsarServiceUrl("pulsar://localhost:6650/"); when(mockedWorkerService.getWorkerConfig()).thenReturn(workerConfig); this.resource = spy(new SourcesImpl(() -> mockedWorkerService)); @@ -199,11 +218,11 @@ public void testRegisterSourceMissingTenant() { mockedFormData, outputTopic, outputSerdeClassName, - className, + TWITTER_FIRE_HOSE, parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -213,18 +232,18 @@ public void testRegisterSourceMissingTenant() { public void testRegisterSourceMissingNamespace() { try { testRegisterSourceMissingArguments( - tenant, - null, - source, - mockedInputStream, - mockedFormData, - outputTopic, - outputSerdeClassName, - className, - parallelism, - null + tenant, + null, + source, + mockedInputStream, + mockedFormData, + outputTopic, + outputSerdeClassName, + TWITTER_FIRE_HOSE, + parallelism, + null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -241,17 +260,18 @@ public void testRegisterSourceMissingSourceName() { mockedFormData, outputTopic, outputSerdeClassName, - className, + TWITTER_FIRE_HOSE, parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source class UnknownClass must be in class path") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source class UnknownClass must" + + " be in class path") public void testRegisterSourceWrongClassName() { try { testRegisterSourceMissingArguments( @@ -266,7 +286,7 @@ public void testRegisterSourceWrongClassName() { parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -287,36 +307,37 @@ public void testRegisterSourceMissingPackage() { parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source Package is not provided") - public void testRegisterSourceMissingPackageDetails() { - try { + public void testRegisterSourceMissingPackageDetails() throws IOException { + try (InputStream inputStream = new FileInputStream(getPulsarIOTwitterNar())) { testRegisterSourceMissingArguments( - tenant, - namespace, + tenant, + namespace, source, - mockedInputStream, - null, - outputTopic, + inputStream, + null, + outputTopic, outputSerdeClassName, - className, - parallelism, + TWITTER_FIRE_HOSE, + parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source package does not have the" + - " correct format. Pulsar cannot determine if the package is a NAR package" + - " or JAR package. Source classname is not provided and attempts to load it as a NAR package produced the following error.") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source package does not have the" + + " correct format. Pulsar cannot determine if the package is a NAR package" + + " or JAR package. Source classname is not provided and attempts to load it as a NAR package " + + "produced the following error.") public void testRegisterSourceMissingPackageDetailsAndClassname() { try { testRegisterSourceMissingArguments( @@ -331,29 +352,29 @@ public void testRegisterSourceMissingPackageDetailsAndClassname() { parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Failed to extract source class from archive") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source package does not have " + + "the correct format.*") public void testRegisterSourceInvalidJarWithNoSource() throws IOException { - try { - FileInputStream inputStream = new FileInputStream(INVALID_JAR_FILE_PATH); + try (InputStream inputStream = new FileInputStream(getPulsarIOInvalidNar())) { testRegisterSourceMissingArguments( tenant, namespace, source, inputStream, - null, + mockedFormData, outputTopic, outputSerdeClassName, null, parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -361,8 +382,7 @@ public void testRegisterSourceInvalidJarWithNoSource() throws IOException { @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Topic name cannot be null") public void testRegisterSourceNoOutputTopic() throws IOException { - try { - FileInputStream inputStream = new FileInputStream(JAR_FILE_PATH); + try (InputStream inputStream = new FileInputStream(getPulsarIOTwitterNar())) { testRegisterSourceMissingArguments( tenant, namespace, @@ -371,17 +391,18 @@ public void testRegisterSourceNoOutputTopic() throws IOException { mockedFormData, null, outputSerdeClassName, - className, + TWITTER_FIRE_HOSE, parallelism, null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Encountered error .*. when getting Source package from .*") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Encountered error .*. when " + + "getting Source package from .*") public void testRegisterSourceHttpUrl() { try { testRegisterSourceMissingArguments( @@ -392,11 +413,11 @@ public void testRegisterSourceHttpUrl() { null, outputTopic, outputSerdeClassName, - className, + TWITTER_FIRE_HOSE, parallelism, "http://localhost:1234/test" ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -476,23 +497,24 @@ public void testUpdateMissingSinkConfig() { } private void registerDefaultSource() throws IOException { - registerDefaultSourceWithPackageUrl(null); + registerDefaultSourceWithPackageUrl("source://public/default/test@v1"); } private void registerDefaultSourceWithPackageUrl(String packageUrl) throws IOException { SourceConfig sourceConfig = createDefaultSourceConfig(); resource.registerSource( - tenant, - namespace, + tenant, + namespace, source, - new FileInputStream(JAR_FILE_PATH), - mockedFormData, - packageUrl, - sourceConfig, + null, + null, + packageUrl, + sourceConfig, null, null); } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source test-source already exists") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source test-source already " + + "exists") public void testRegisterExistedSource() throws IOException { try { Configurator.setRootLevel(Level.DEBUG); @@ -500,7 +522,7 @@ public void testRegisterExistedSource() throws IOException { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(true); registerDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -519,9 +541,10 @@ public void testRegisterSourceUploadFailure() throws Exception { PowerMockito.when(WorkerUtils.class, "dumpToTmpFile", any()).thenCallRealMethod(); when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(false); + when(mockedRuntimeFactory.externallyManaged()).thenReturn(true); registerDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.INTERNAL_SERVER_ERROR); throw re; } @@ -552,7 +575,7 @@ public void testRegisterSourceSuccessWithPackageName() throws IOException { public void testRegisterSourceFailedWithWrongPackageName() throws PulsarAdminException, IOException { try { doThrow(new PulsarAdminException("package name is invalid")) - .when(mockedPackages).download(anyString(), anyString()); + .when(mockedPackages).download(anyString(), anyString()); registerDefaultSourceWithPackageUrl("source://"); } catch (RestException e) { // expected exception @@ -582,19 +605,21 @@ public void testRegisterSourceConflictingFields() throws Exception { sourceConfig.setTenant(tenant); sourceConfig.setNamespace(namespace); sourceConfig.setName(source); - sourceConfig.setClassName(className); + sourceConfig.setClassName(TWITTER_FIRE_HOSE); sourceConfig.setParallelism(parallelism); sourceConfig.setTopicName(outputTopic); sourceConfig.setSerdeClassName(outputSerdeClassName); - resource.registerSource( - actualTenant, - actualNamespace, - actualName, - new FileInputStream(JAR_FILE_PATH), - mockedFormData, - null, - sourceConfig, - null, null); + try (InputStream inputStream = new FileInputStream(getPulsarIOTwitterNar())) { + resource.registerSource( + actualTenant, + actualNamespace, + actualName, + inputStream, + mockedFormData, + null, + sourceConfig, + null, null); + } } @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "source failed to register") @@ -615,13 +640,14 @@ public void testRegisterSourceFailure() throws Exception { .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); registerDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function registration interrupted") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function registration " + + "interrupted") public void testRegisterSourceInterrupted() throws Exception { try { mockStatic(WorkerUtils.class); @@ -639,7 +665,7 @@ public void testRegisterSourceInterrupted() throws Exception { .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); registerDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.INTERNAL_SERVER_ERROR); throw re; } @@ -653,17 +679,17 @@ public void testRegisterSourceInterrupted() throws Exception { public void testUpdateSourceMissingTenant() throws Exception { try { testUpdateSourceMissingArguments( - null, - namespace, + null, + namespace, source, - mockedInputStream, - mockedFormData, - outputTopic, + mockedInputStream, + mockedFormData, + outputTopic, outputSerdeClassName, - className, - parallelism, + TWITTER_FIRE_HOSE, + parallelism, "Tenant is not provided"); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -673,17 +699,17 @@ public void testUpdateSourceMissingTenant() throws Exception { public void testUpdateSourceMissingNamespace() throws Exception { try { testUpdateSourceMissingArguments( - tenant, - null, + tenant, + null, source, - mockedInputStream, - mockedFormData, - outputTopic, + mockedInputStream, + mockedFormData, + outputTopic, outputSerdeClassName, - className, - parallelism, + TWITTER_FIRE_HOSE, + parallelism, "Namespace is not provided"); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -693,17 +719,17 @@ public void testUpdateSourceMissingNamespace() throws Exception { public void testUpdateSourceMissingFunctionName() throws Exception { try { testUpdateSourceMissingArguments( - tenant, - namespace, - null, - mockedInputStream, - mockedFormData, - outputTopic, + tenant, + namespace, + null, + mockedInputStream, + mockedFormData, + outputTopic, outputSerdeClassName, - className, - parallelism, + TWITTER_FIRE_HOSE, + parallelism, "Source name is not provided"); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -717,17 +743,17 @@ public void testUpdateSourceMissingPackage() throws Exception { WorkerUtils.downloadFromBookkeeper(any(Namespace.class), any(File.class), anyString()); testUpdateSourceMissingArguments( - tenant, - namespace, + tenant, + namespace, source, null, - mockedFormData, - outputTopic, + mockedFormData, + outputTopic, outputSerdeClassName, - null, - parallelism, + null, + parallelism, "Update contains no change"); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -751,13 +777,14 @@ public void testUpdateSourceMissingTopicName() throws Exception { null, parallelism, "Update contains no change"); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source parallelism must be a positive number") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source parallelism must be a " + + "positive number") public void testUpdateSourceNegativeParallelism() throws Exception { try { mockStatic(WorkerUtils.class); @@ -774,10 +801,10 @@ public void testUpdateSourceNegativeParallelism() throws Exception { mockedFormData, outputTopic, outputSerdeClassName, - className, + TWITTER_FIRE_HOSE, -2, "Source parallelism must be a positive number"); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -793,17 +820,17 @@ public void testUpdateSourceChangedParallelism() throws Exception { PowerMockito.when(WorkerUtils.class, "dumpToTmpFile", any()).thenCallRealMethod(); testUpdateSourceMissingArguments( - tenant, - namespace, - source, - null, - mockedFormData, - outputTopic, - outputSerdeClassName, - className, - parallelism + 1, - null); - } catch (RestException re){ + tenant, + namespace, + source, + null, + mockedFormData, + outputTopic, + outputSerdeClassName, + TWITTER_FIRE_HOSE, + parallelism + 1, + null); + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -823,12 +850,13 @@ public void testUpdateSourceChangedTopic() throws Exception { mockedFormData, "DifferentTopic", outputSerdeClassName, - className, + TWITTER_FIRE_HOSE, parallelism, null); } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source parallelism must be a positive number") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source parallelism must be a " + + "positive number") public void testUpdateSourceZeroParallelism() throws Exception { try { mockStatic(WorkerUtils.class); @@ -845,10 +873,10 @@ public void testUpdateSourceZeroParallelism() throws Exception { mockedFormData, outputTopic, outputSerdeClassName, - className, + TWITTER_FIRE_HOSE, 0, "Source parallelism must be a positive number"); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -866,26 +894,25 @@ private void testUpdateSourceMissingArguments( Integer parallelism, String expectedError) throws Exception { - NarClassLoader classLoader = mock(NarClassLoader.class); - doReturn(TwitterFireHose.class).when(classLoader).loadClass(eq(TwitterFireHose.class.getName())); - mockStatic(ConnectorUtils.class); - doReturn(TwitterFireHose.class.getName()).when(ConnectorUtils.class); + doReturn(TWITTER_FIRE_HOSE).when(ConnectorUtils.class); ConnectorUtils.getIOSourceClass(any(NarClassLoader.class)); mockStatic(ClassLoaderUtils.class); mockStatic(FunctionCommon.class); PowerMockito.when(FunctionCommon.class, "createPkgTempFile").thenCallRealMethod(); - PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()).thenCallRealMethod(); + PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()) + .thenCallRealMethod(); doReturn(String.class).when(FunctionCommon.class); - FunctionCommon.getSourceType(eq(TwitterFireHose.class)); + FunctionCommon.getSourceType(argThat(clazz -> clazz.getName().equals(TWITTER_FIRE_HOSE))); - doReturn(classLoader).when(FunctionCommon.class); + doReturn(narClassLoader).when(FunctionCommon.class); FunctionCommon.extractNarClassLoader(any(), any()); - this.mockedFunctionMetaData = FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); + this.mockedFunctionMetaData = + FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); when(mockedManager.getFunctionMetaData(any(), any(), any())).thenReturn(mockedFunctionMetaData); when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(function))).thenReturn(true); @@ -919,13 +946,13 @@ private void testUpdateSourceMissingArguments( } resource.updateSource( - tenant, - namespace, - function, - inputStream, - details, - null, - sourceConfig, + tenant, + namespace, + function, + inputStream, + details, + null, + sourceConfig, null, null, null); } @@ -939,50 +966,52 @@ private void updateDefaultSourceWithPackageUrl(String packageUrl) throws Excepti sourceConfig.setTenant(tenant); sourceConfig.setNamespace(namespace); sourceConfig.setName(source); - sourceConfig.setClassName(className); + sourceConfig.setClassName(TWITTER_FIRE_HOSE); sourceConfig.setParallelism(parallelism); sourceConfig.setTopicName(outputTopic); sourceConfig.setSerdeClassName(outputSerdeClassName); - NarClassLoader classLoader = mock(NarClassLoader.class); - doReturn(TwitterFireHose.class).when(classLoader).loadClass(eq(TwitterFireHose.class.getName())); - mockStatic(ConnectorUtils.class); - doReturn(TwitterFireHose.class.getName()).when(ConnectorUtils.class); + doReturn(TWITTER_FIRE_HOSE).when(ConnectorUtils.class); ConnectorUtils.getIOSourceClass(any(NarClassLoader.class)); mockStatic(ClassLoaderUtils.class); mockStatic(FunctionCommon.class); PowerMockito.when(FunctionCommon.class, "createPkgTempFile").thenCallRealMethod(); - PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()).thenCallRealMethod(); + PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()) + .thenCallRealMethod(); doReturn(String.class).when(FunctionCommon.class); - FunctionCommon.getSourceType(eq(TwitterFireHose.class)); + FunctionCommon.getSourceType(argThat(clazz -> clazz.getName().equals(TWITTER_FIRE_HOSE))); - doReturn(classLoader).when(FunctionCommon.class); + doReturn(narClassLoader).when(FunctionCommon.class); FunctionCommon.extractNarClassLoader(any(File.class), any()); - this.mockedFunctionMetaData = FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); + this.mockedFunctionMetaData = + FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); when(mockedManager.getFunctionMetaData(any(), any(), any())).thenReturn(mockedFunctionMetaData); - resource.updateSource( - tenant, - namespace, - source, - new FileInputStream(JAR_FILE_PATH), - mockedFormData, - packageUrl, - sourceConfig, - null, null, null); + try (InputStream inputStream = new FileInputStream(getPulsarIOCassandraNar())) { + resource.updateSource( + tenant, + namespace, + source, + inputStream, + mockedFormData, + packageUrl, + sourceConfig, + null, null, null); + } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source test-source doesn't exist") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source test-source doesn't " + + "exist") public void testUpdateNotExistedSource() throws Exception { try { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(false); updateDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1002,7 +1031,7 @@ public void testUpdateSourceUploadFailure() throws Exception { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(true); updateDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.INTERNAL_SERVER_ERROR); throw re; } @@ -1028,7 +1057,7 @@ public void testUpdateSourceSuccess() throws Exception { public void testUpdateSourceWithUrl() throws Exception { Configurator.setRootLevel(Level.DEBUG); - String filePackageUrl = "file://" + JAR_FILE_PATH; + String filePackageUrl = getPulsarIOCassandraNar().toURI().toString(); SourceConfig sourceConfig = new SourceConfig(); sourceConfig.setTopicName(outputTopic); @@ -1036,39 +1065,38 @@ public void testUpdateSourceWithUrl() throws Exception { sourceConfig.setTenant(tenant); sourceConfig.setNamespace(namespace); sourceConfig.setName(source); - sourceConfig.setClassName(className); + sourceConfig.setClassName(TWITTER_FIRE_HOSE); sourceConfig.setParallelism(parallelism); - NarClassLoader classLoader = mock(NarClassLoader.class); - doReturn(TwitterFireHose.class).when(classLoader).loadClass(eq(TwitterFireHose.class.getName())); - when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(true); mockStatic(ConnectorUtils.class); - doReturn(TwitterFireHose.class.getName()).when(ConnectorUtils.class); + doReturn(TWITTER_FIRE_HOSE).when(ConnectorUtils.class); ConnectorUtils.getIOSourceClass(any(NarClassLoader.class)); mockStatic(ClassLoaderUtils.class); mockStatic(FunctionCommon.class); doReturn(String.class).when(FunctionCommon.class); - FunctionCommon.getSourceType(eq(TwitterFireHose.class)); + FunctionCommon.getSourceType(argThat(clazz -> clazz.getName().equals(TWITTER_FIRE_HOSE))); PowerMockito.when(FunctionCommon.class, "extractFileFromPkgURL", any()).thenCallRealMethod(); - PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()).thenCallRealMethod(); + PowerMockito.when(FunctionCommon.class, "getClassLoaderFromPackage", any(), any(), any(), any()) + .thenCallRealMethod(); - doReturn(classLoader).when(FunctionCommon.class); + doReturn(narClassLoader).when(FunctionCommon.class); FunctionCommon.extractNarClassLoader(any(), any()); - this.mockedFunctionMetaData = FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); + this.mockedFunctionMetaData = + FunctionMetaData.newBuilder().setFunctionDetails(createDefaultFunctionDetails()).build(); when(mockedManager.getFunctionMetaData(any(), any(), any())).thenReturn(mockedFunctionMetaData); resource.updateSource( - tenant, - namespace, + tenant, + namespace, source, - null, - null, - filePackageUrl, - sourceConfig, + null, + null, + filePackageUrl, + sourceConfig, null, null, null); } @@ -1091,13 +1119,14 @@ public void testUpdateSourceFailure() throws Exception { .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); updateDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function registration interrupted") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function registration " + + "interrupted") public void testUpdateSourceInterrupted() throws Exception { try { mockStatic(WorkerUtils.class); @@ -1115,7 +1144,7 @@ public void testUpdateSourceInterrupted() throws Exception { .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); updateDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.INTERNAL_SERVER_ERROR); throw re; } @@ -1132,7 +1161,7 @@ public void testUpdateSourceFailedWithWrongPackageName() throws Exception { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(true); try { doThrow(new PulsarAdminException("package name is invalid")) - .when(mockedPackages).download(anyString(), anyString()); + .when(mockedPackages).download(anyString(), anyString()); updateDefaultSourceWithPackageUrl("source://"); } catch (RestException e) { // expected exception @@ -1148,11 +1177,11 @@ public void testUpdateSourceFailedWithWrongPackageName() throws Exception { public void testDeregisterSourceMissingTenant() { try { testDeregisterSourceMissingArguments( - null, - namespace, + null, + namespace, source ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1162,11 +1191,11 @@ public void testDeregisterSourceMissingTenant() { public void testDeregisterSourceMissingNamespace() { try { testDeregisterSourceMissingArguments( - tenant, - null, - source + tenant, + null, + source ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1176,11 +1205,11 @@ public void testDeregisterSourceMissingNamespace() { public void testDeregisterSourceMissingFunctionName() { try { testDeregisterSourceMissingArguments( - tenant, - namespace, - null + tenant, + namespace, + null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1192,27 +1221,28 @@ private void testDeregisterSourceMissingArguments( String function ) { resource.deregisterFunction( - tenant, - namespace, - function, + tenant, + namespace, + function, null, null); } private void deregisterDefaultSource() { resource.deregisterFunction( - tenant, - namespace, + tenant, + namespace, source, null, null); } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp= "Source test-source doesn't exist") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source test-source doesn't " + + "exist") public void testDeregisterNotExistedSource() { try { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(false); deregisterDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.NOT_FOUND); throw re; } @@ -1222,7 +1252,8 @@ public void testDeregisterNotExistedSource() { public void testDeregisterSourceSuccess() { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(true); - when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(source))).thenReturn(FunctionMetaData.newBuilder().build()); + when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(source))) + .thenReturn(FunctionMetaData.newBuilder().build()); deregisterDefaultSource(); } @@ -1232,30 +1263,33 @@ public void testDeregisterSourceFailure() throws Exception { try { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(true); - when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(source))).thenReturn(FunctionMetaData.newBuilder().build()); + when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(source))) + .thenReturn(FunctionMetaData.newBuilder().build()); doThrow(new IllegalArgumentException("source failed to deregister")) .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); deregisterDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function deregistration interrupted") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function deregistration " + + "interrupted") public void testDeregisterSourceInterrupted() throws Exception { try { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(true); - when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(source))).thenReturn(FunctionMetaData.newBuilder().build()); + when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(source))) + .thenReturn(FunctionMetaData.newBuilder().build()); doThrow(new IllegalStateException("Function deregistration interrupted")) .when(mockedManager).updateFunctionOnLeader(any(FunctionMetaData.class), Mockito.anyBoolean()); deregisterDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.INTERNAL_SERVER_ERROR); throw re; } @@ -1268,7 +1302,8 @@ public void testDeregisterSourceBKPackageCleanup() throws IOException { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(true); - String packagePath = "public/default/test/591541f0-c7c5-40c0-983b-610c722f90b0-pulsar-io-batch-data-generator-2.7.0.nar"; + String packagePath = + "public/default/test/591541f0-c7c5-40c0-983b-610c722f90b0-pulsar-io-batch-data-generator-2.7.0.nar"; when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(source))) .thenReturn(FunctionMetaData.newBuilder().setPackageLocation( PackageLocationMetaData.newBuilder().setPackagePath(packagePath).build()).build()); @@ -1348,7 +1383,7 @@ public void testGetSourceMissingTenant() { namespace, source ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1358,11 +1393,11 @@ public void testGetSourceMissingTenant() { public void testGetSourceMissingNamespace() { try { testGetSourceMissingArguments( - tenant, - null, + tenant, + null, source ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1372,11 +1407,11 @@ public void testGetSourceMissingNamespace() { public void testGetSourceMissingFunctionName() { try { testGetSourceMissingArguments( - tenant, - namespace, - null + tenant, + namespace, + null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1388,26 +1423,27 @@ private void testGetSourceMissingArguments( String source ) { resource.getFunctionInfo( - tenant, - namespace, - source, null, null + tenant, + namespace, + source, null, null ); } private SourceConfig getDefaultSourceInfo() { return resource.getSourceInfo( - tenant, - namespace, + tenant, + namespace, source ); } - @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source test-source doesn't exist") + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Source test-source doesn't " + + "exist") public void testGetNotExistedSource() { try { when(mockedManager.containsFunction(eq(tenant), eq(namespace), eq(source))).thenReturn(false); getDefaultSourceInfo(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.NOT_FOUND); throw re; } @@ -1433,11 +1469,11 @@ public void testGetSourceSuccess() { .setParallelism(parallelism) .setSource(sourceSpec).build(); FunctionMetaData metaData = FunctionMetaData.newBuilder() - .setCreateTime(System.currentTimeMillis()) - .setFunctionDetails(functionDetails) - .setPackageLocation(PackageLocationMetaData.newBuilder().setPackagePath("/path/to/package")) - .setVersion(1234) - .build(); + .setCreateTime(System.currentTimeMillis()) + .setFunctionDetails(functionDetails) + .setPackageLocation(PackageLocationMetaData.newBuilder().setPackagePath("/path/to/package")) + .setVersion(1234) + .build(); when(mockedManager.getFunctionMetaData(eq(tenant), eq(namespace), eq(source))).thenReturn(metaData); SourceConfig config = getDefaultSourceInfo(); @@ -1452,10 +1488,10 @@ public void testGetSourceSuccess() { public void testListSourcesMissingTenant() { try { testListSourcesMissingArguments( - null, - namespace + null, + namespace ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1465,10 +1501,10 @@ public void testListSourcesMissingTenant() { public void testListSourcesMissingNamespace() { try { testListSourcesMissingArguments( - tenant, - null + tenant, + null ); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1479,15 +1515,15 @@ private void testListSourcesMissingArguments( String namespace ) { resource.listFunctions( - tenant, - namespace, null, null + tenant, + namespace, null, null ); } private List listDefaultSources() { return resource.listFunctions( - tenant, - namespace, null, null); + tenant, + namespace, null, null); } @Test @@ -1521,9 +1557,12 @@ public void testOnlyGetSources() { functionMetaDataList.add(f3); when(mockedManager.listFunctions(eq(tenant), eq(namespace))).thenReturn(functionMetaDataList); mockStatic(InstanceUtils.class); - PowerMockito.when(InstanceUtils.calculateSubjectType(f1.getFunctionDetails())).thenReturn(FunctionDetails.ComponentType.SOURCE); - PowerMockito.when(InstanceUtils.calculateSubjectType(f2.getFunctionDetails())).thenReturn(FunctionDetails.ComponentType.FUNCTION); - PowerMockito.when(InstanceUtils.calculateSubjectType(f3.getFunctionDetails())).thenReturn(FunctionDetails.ComponentType.SINK); + PowerMockito.when(InstanceUtils.calculateSubjectType(f1.getFunctionDetails())) + .thenReturn(FunctionDetails.ComponentType.SOURCE); + PowerMockito.when(InstanceUtils.calculateSubjectType(f2.getFunctionDetails())) + .thenReturn(FunctionDetails.ComponentType.FUNCTION); + PowerMockito.when(InstanceUtils.calculateSubjectType(f3.getFunctionDetails())) + .thenReturn(FunctionDetails.ComponentType.SINK); List sourceList = listDefaultSources(); assertEquals(functions, sourceList); @@ -1534,7 +1573,7 @@ public void testRegisterFunctionNonExistingNamespace() throws Exception { try { this.namespaceList.clear(); registerDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1545,7 +1584,7 @@ public void testRegisterFunctionNonExistingTenant() throws Exception { try { when(mockedTenants.getTenantInfo(any())).thenThrow(PulsarAdminException.NotFoundException.class); registerDefaultSource(); - } catch (RestException re){ + } catch (RestException re) { assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); throw re; } @@ -1556,7 +1595,7 @@ private SourceConfig createDefaultSourceConfig() { sourceConfig.setTenant(tenant); sourceConfig.setNamespace(namespace); sourceConfig.setName(source); - sourceConfig.setClassName(className); + sourceConfig.setClassName(TWITTER_FIRE_HOSE); sourceConfig.setParallelism(parallelism); sourceConfig.setTopicName(outputTopic); sourceConfig.setSerdeClassName(outputSerdeClassName);