You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/07/31 18:32:51 UTC
[GitHub] rdhabalia closed pull request #2258: Derive source/sink arg-class
name from function-class for file-url
rdhabalia closed pull request #2258: Derive source/sink arg-class name from function-class for file-url
URL: https://github.com/apache/incubator-pulsar/pull/2258
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
index 16f1a76597..5398bc9b15 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
@@ -21,6 +21,10 @@
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
import static org.mockito.Mockito.spy;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
import java.io.File;
import java.lang.reflect.Method;
@@ -62,7 +66,6 @@
import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatusList;
-import org.apache.pulsar.functions.proto.InstanceCommunication.MetricsData;
import org.apache.pulsar.functions.proto.InstanceCommunication.MetricsData.DataDigest;
import org.apache.pulsar.functions.sink.PulsarSink;
import org.apache.pulsar.functions.utils.Reflections;
@@ -70,11 +73,9 @@
import org.apache.pulsar.functions.worker.FunctionRuntimeManager;
import org.apache.pulsar.functions.worker.WorkerConfig;
import org.apache.pulsar.functions.worker.WorkerService;
-import org.apache.pulsar.functions.worker.rest.WorkerServer;
import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
@@ -124,7 +125,7 @@
public Object[][] validRoleName() {
return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
}
-
+
@BeforeMethod
void setup(Method method) throws Exception {
@@ -147,7 +148,6 @@ void setup(Method method) throws Exception {
config.setBrokerServicePortTls(brokerServiceTlsPort);
config.setLoadManagerClassName(SimpleLoadManagerImpl.class.getName());
-
Set<String> providers = new HashSet<>();
providers.add(AuthenticationProviderTls.class.getName());
config.setAuthenticationEnabled(true);
@@ -156,7 +156,6 @@ void setup(Method method) throws Exception {
config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
config.setTlsAllowInsecureConnection(true);
-
functionsWorkerService = createPulsarFunctionWorker(config);
urlTls = new URL(brokerServiceUrl);
@@ -190,12 +189,12 @@ void setup(Method method) throws Exception {
workerConfig.getClientAuthenticationParameters());
}
pulsarClient = clientBuilder.build();
-
+
TenantInfo propAdmin = new TenantInfo();
propAdmin.getAdminRoles().add("superUser");
propAdmin.setAllowedClusters(Sets.newHashSet(Lists.newArrayList("use")));
admin.tenants().updateTenant(tenant, propAdmin);
-
+
Thread.sleep(100);
}
@@ -237,7 +236,7 @@ private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) {
workerConfig.setUseTls(true);
workerConfig.setTlsAllowInsecureConnection(true);
workerConfig.setTlsTrustCertsFilePath(TLS_CLIENT_CERT_FILE_PATH);
-
+
workerConfig.setAuthenticationEnabled(true);
workerConfig.setAuthorizationEnabled(true);
@@ -285,7 +284,7 @@ public void testE2EPulsarSink() throws Exception {
}
}, 5, 150);
// validate pulsar sink consumer has started on the topic
- Assert.assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);
+ assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);
int totalMsgs = 5;
for (int i = 0; i < totalMsgs; i++) {
@@ -303,17 +302,15 @@ public void testE2EPulsarSink() throws Exception {
Message<byte[]> msg = consumer.receive(5, TimeUnit.SECONDS);
String receivedPropertyValue = msg.getProperty(propertyKey);
- Assert.assertEquals(propertyValue, receivedPropertyValue);
+ assertEquals(propertyValue, receivedPropertyValue);
// validate pulsar-sink consumer has consumed all messages and delivered to Pulsar sink but unacked messages
// due to publish failure
- Assert.assertNotEquals(
- admin.topics().getStats(sourceTopic).subscriptions.values().iterator().next().unackedMessages,
+ assertNotEquals(admin.topics().getStats(sourceTopic).subscriptions.values().iterator().next().unackedMessages,
totalMsgs);
}
-
@Test(timeOut = 20000)
public void testPulsarSinkStats() throws Exception {
@@ -349,7 +346,7 @@ public void testPulsarSinkStats() throws Exception {
}
}, 5, 150);
// validate pulsar sink consumer has started on the topic
- Assert.assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);
+ assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);
int totalMsgs = 10;
for (int i = 0; i < totalMsgs; i++) {
@@ -371,7 +368,7 @@ public void testPulsarSinkStats() throws Exception {
functionName);
int numInstances = functionStats.getFunctionStatusListCount();
- Assert.assertEquals(numInstances, 1);
+ assertEquals(numInstances, 1);
FunctionStatus stats = functionStats.getFunctionStatusListList().get(0);
Map<String, DataDigest> metricsData = stats.getMetrics().getMetricsMap();
@@ -379,12 +376,13 @@ public void testPulsarSinkStats() throws Exception {
double count = metricsData.get(JavaInstanceRunnable.METRICS_TOTAL_PROCESSED).getCount();
double success = metricsData.get(JavaInstanceRunnable.METRICS_TOTAL_SUCCESS).getCount();
String ownerWorkerId = stats.getWorkerId();
- Assert.assertEquals((int) count, totalMsgs);
- Assert.assertEquals((int) success, totalMsgs);
- Assert.assertEquals(ownerWorkerId, workerId);
+ assertEquals((int) count, totalMsgs);
+ assertEquals((int) success, totalMsgs);
+ assertEquals(ownerWorkerId, workerId);
}
- protected FunctionDetails createSinkConfig(String jarFile, String tenant, String namespace, String functionName, String sinkTopic, String subscriptionName) {
+ protected FunctionDetails createSinkConfig(String jarFile, String tenant, String namespace, String functionName,
+ String sinkTopic, String subscriptionName) {
File file = new File(jarFile);
try {
@@ -407,7 +405,7 @@ protected FunctionDetails createSinkConfig(String jarFile, String tenant, String
// source spec classname should be empty so that the default pulsar source will be used
SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.SHARED);
- sourceSpecBuilder.setTypeClassName(byte[].class.getName());
+ sourceSpecBuilder.setTypeClassName(typeArg.getName());
sourceSpecBuilder.setTopicsPattern(sourceTopicPattern);
sourceSpecBuilder.setSubscriptionName(subscriptionName);
sourceSpecBuilder.putTopicsToSerDeClassName(sourceTopicPattern, DefaultSerDe.class.getName());
@@ -425,7 +423,7 @@ protected FunctionDetails createSinkConfig(String jarFile, String tenant, String
return functionDetailsBuilder.build();
}
-
+
@Test(dataProvider = "validRoleName")
public void testAuthorization(boolean validRoleName) throws Exception {
@@ -450,9 +448,62 @@ public void testAuthorization(boolean validRoleName) throws Exception {
sinkTopic, subscriptionName);
try {
admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
- Assert.assertTrue(validRoleName);
+ assertTrue(validRoleName);
} catch (org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException ne) {
- Assert.assertFalse(validRoleName);
+ assertFalse(validRoleName);
}
}
+
+ /**
+ * Test to verify: function-server loads jar using file-url and derives type-args classes if not provided
+ * @throws Exception
+ */
+ @Test(timeOut = 20000)
+ public void testFileUrlFunctionWithoutPassingTypeArgs() throws Exception {
+
+ final String namespacePortion = "io";
+ final String replNamespace = tenant + "/" + namespacePortion;
+ final String sinkTopic = "persistent://" + replNamespace + "/output";
+ final String functionName = "PulsarSink-test";
+ admin.namespaces().createNamespace(replNamespace);
+ Set<String> clusters = Sets.newHashSet(Lists.newArrayList("use"));
+ admin.namespaces().setNamespaceReplicationClusters(replNamespace, clusters);
+
+ String jarFilePathUrl = Utils.FILE + ":"
+ + IdentityFunction.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+
+ FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
+ functionDetailsBuilder.setTenant(tenant);
+ functionDetailsBuilder.setNamespace(namespacePortion);
+ functionDetailsBuilder.setName(functionName);
+ functionDetailsBuilder.setRuntime(FunctionDetails.Runtime.JAVA);
+ functionDetailsBuilder.setParallelism(1);
+ functionDetailsBuilder.setClassName(IdentityFunction.class.getName());
+
+ Class<?>[] typeArgs = org.apache.pulsar.functions.utils.Utils.getFunctionTypes(new IdentityFunction(), false);
+
+ // set source spec
+ // source spec classname should be empty so that the default pulsar source will be used
+ SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
+ sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.SHARED);
+ sourceSpecBuilder.putTopicsToSerDeClassName(sinkTopic, DefaultSerDe.class.getName());
+ functionDetailsBuilder.setAutoAck(true);
+ functionDetailsBuilder.setSource(sourceSpecBuilder);
+
+ // set up sink spec
+ SinkSpec.Builder sinkSpecBuilder = SinkSpec.newBuilder();
+ sinkSpecBuilder.setTopic(sinkTopic);
+ Map<String, Object> sinkConfigMap = Maps.newHashMap();
+ sinkSpecBuilder.setConfigs(new Gson().toJson(sinkConfigMap));
+ functionDetailsBuilder.setSink(sinkSpecBuilder);
+
+ FunctionDetails functionDetails = functionDetailsBuilder.build();
+ admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+
+ FunctionDetails functionMetadata = admin.functions().getFunction(tenant, namespacePortion, functionName);
+
+ assertEquals(functionMetadata.getSource().getTypeClassName(), typeArgs[0].getName());
+ assertEquals(functionMetadata.getSink().getTypeClassName(), typeArgs[1].getName());
+
+ }
}
\ No newline at end of file
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Utils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Utils.java
index 0c25be29ca..94c315df3f 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Utils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Utils.java
@@ -102,16 +102,21 @@ public static int findAvailablePort() {
}
public static Class<?>[] getFunctionTypes(FunctionConfig functionConfig) {
-
- Object userClass = createInstance(functionConfig.getClassName(), Thread.currentThread().getContextClassLoader());
+ Object userClass = createInstance(functionConfig.getClassName(),
+ Thread.currentThread().getContextClassLoader());
+ boolean isWindowConfigPresent = functionConfig.getWindowConfig() != null;
+ return getFunctionTypes(userClass, isWindowConfigPresent);
+ }
+
+ public static Class<?>[] getFunctionTypes(Object userClass, boolean isWindowConfigPresent) {
Class<?>[] typeArgs;
// if window function
- if (functionConfig.getWindowConfig() != null) {
+ if (isWindowConfigPresent) {
java.util.function.Function function = (java.util.function.Function) userClass;
if (function == null) {
- throw new IllegalArgumentException(String.format("The Java util function class %s could not be instantiated",
- functionConfig.getClassName()));
+ throw new IllegalArgumentException(
+ String.format("The Java util function class %s could not be instantiated", userClass));
}
typeArgs = TypeResolver.resolveRawArguments(java.util.function.Function.class, function.getClass());
if (!typeArgs[0].equals(Collection.class)) {
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
index 9b0ad98b27..989d09026b 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
@@ -56,7 +56,10 @@
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang3.StringUtils;
+import static org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.commons.lang3.StringUtils.isEmpty;
+import static org.apache.commons.lang3.StringUtils.join;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
@@ -139,7 +142,7 @@ public Response registerFunction(final String tenant, final String namespace, fi
}
FunctionDetails functionDetails;
- boolean isPkgUrlProvided = StringUtils.isNotBlank(functionPkgUrl);
+ boolean isPkgUrlProvided = isNotBlank(functionPkgUrl);
// validate parameters
try {
if (isPkgUrlProvided) {
@@ -203,7 +206,7 @@ public Response updateFunction(final String tenant, final String namespace, fina
}
FunctionDetails functionDetails;
- boolean isPkgUrlProvided = StringUtils.isNotBlank(functionPkgUrl);
+ boolean isPkgUrlProvided = isNotBlank(functionPkgUrl);
// validate parameters
try {
if (isPkgUrlProvided) {
@@ -756,14 +759,14 @@ private FunctionDetails validateUpdateRequestParams(String tenant, String namesp
private boolean isFunctionCodeBuiltin(FunctionDetails functionDetails) {
if (functionDetails.hasSource()) {
SourceSpec sourceSpec = functionDetails.getSource();
- if (!StringUtils.isEmpty(sourceSpec.getBuiltin())) {
+ if (!isEmpty(sourceSpec.getBuiltin())) {
return true;
}
}
if (functionDetails.hasSink()) {
SinkSpec sinkSpec = functionDetails.getSink();
- if (!StringUtils.isEmpty(sinkSpec.getBuiltin())) {
+ if (!isEmpty(sinkSpec.getBuiltin())) {
return true;
}
}
@@ -774,14 +777,14 @@ private boolean isFunctionCodeBuiltin(FunctionDetails functionDetails) {
private String getFunctionCodeBuiltin(FunctionDetails functionDetails) {
if (functionDetails.hasSource()) {
SourceSpec sourceSpec = functionDetails.getSource();
- if (!StringUtils.isEmpty(sourceSpec.getBuiltin())) {
+ if (!isEmpty(sourceSpec.getBuiltin())) {
return sourceSpec.getBuiltin();
}
}
if (functionDetails.hasSink()) {
SinkSpec sinkSpec = functionDetails.getSink();
- if (!StringUtils.isEmpty(sinkSpec.getBuiltin())) {
+ if (!isEmpty(sinkSpec.getBuiltin())) {
return sinkSpec.getBuiltin();
}
}
@@ -833,7 +836,7 @@ private FunctionDetails validateUpdateRequestParams(String tenant, String namesp
missingFields.add("Sink");
}
if (!missingFields.isEmpty()) {
- String errorMessage = StringUtils.join(missingFields, ",");
+ String errorMessage = join(missingFields, ",");
throw new IllegalArgumentException(errorMessage + " is not provided");
}
if (functionDetails.getParallelism() <= 0) {
@@ -855,7 +858,7 @@ private void validateFunctionClassTypes(File jarFile, FunctionDetails.Builder fu
return;
}
- if (StringUtils.isBlank(functionDetailsBuilder.getClassName())) {
+ if (isBlank(functionDetailsBuilder.getClassName())) {
throw new IllegalArgumentException("function class-name can't be empty");
}
@@ -865,13 +868,15 @@ private void validateFunctionClassTypes(File jarFile, FunctionDetails.Builder fu
// validate function class-type
Object functionObject = createInstance(functionDetailsBuilder.getClassName(), classLoader);
+ Class<?>[] typeArgs = org.apache.pulsar.functions.utils.Utils.getFunctionTypes(functionObject, false);
+
if (!(functionObject instanceof org.apache.pulsar.functions.api.Function)
&& !(functionObject instanceof java.util.function.Function)) {
throw new RuntimeException("User class must either be Function or java.util.Function");
}
-
+
if (functionDetailsBuilder.hasSource() && functionDetailsBuilder.getSource() != null
- && StringUtils.isNotBlank(functionDetailsBuilder.getSource().getClassName())) {
+ && isNotBlank(functionDetailsBuilder.getSource().getClassName())) {
try {
String sourceClassName = functionDetailsBuilder.getSource().getClassName();
String argClassName = getTypeArg(sourceClassName, Source.class, classLoader).getName();
@@ -880,7 +885,7 @@ private void validateFunctionClassTypes(File jarFile, FunctionDetails.Builder fu
// if sink-class not present then set same arg as source
if (!functionDetailsBuilder.hasSink()
- || StringUtils.isBlank(functionDetailsBuilder.getSink().getClassName())) {
+ || isBlank(functionDetailsBuilder.getSink().getClassName())) {
functionDetailsBuilder
.setSink(functionDetailsBuilder.getSinkBuilder().setTypeClassName(argClassName));
}
@@ -891,10 +896,14 @@ private void validateFunctionClassTypes(File jarFile, FunctionDetails.Builder fu
log.error("Failed to validate source class", e);
throw new IllegalArgumentException("Failed to validate source class-name", e);
}
+ } else if (isBlank(functionDetailsBuilder.getSourceBuilder().getTypeClassName())) {
+ // if function-src-class is not present then set function-src type-class according to function class
+ functionDetailsBuilder
+ .setSource(functionDetailsBuilder.getSourceBuilder().setTypeClassName(typeArgs[0].getName()));
}
if (functionDetailsBuilder.hasSink() && functionDetailsBuilder.getSink() != null
- && StringUtils.isNotBlank(functionDetailsBuilder.getSink().getClassName())) {
+ && isNotBlank(functionDetailsBuilder.getSink().getClassName())) {
try {
String sinkClassName = functionDetailsBuilder.getSink().getClassName();
String argClassName = getTypeArg(sinkClassName, Sink.class, classLoader).getName();
@@ -902,7 +911,7 @@ private void validateFunctionClassTypes(File jarFile, FunctionDetails.Builder fu
// if source-class not present then set same arg as sink
if (!functionDetailsBuilder.hasSource()
- || StringUtils.isBlank(functionDetailsBuilder.getSource().getClassName())) {
+ || isBlank(functionDetailsBuilder.getSource().getClassName())) {
functionDetailsBuilder
.setSource(functionDetailsBuilder.getSourceBuilder().setTypeClassName(argClassName));
}
@@ -913,7 +922,12 @@ private void validateFunctionClassTypes(File jarFile, FunctionDetails.Builder fu
log.error("Failed to validate sink class", e);
throw new IllegalArgumentException("Failed to validate sink class-name", e);
}
+ } else if(isBlank(functionDetailsBuilder.getSinkBuilder().getTypeClassName())){
+ // if function-sink-class is not present then set function-sink type-class according to function class
+ functionDetailsBuilder
+ .setSink(functionDetailsBuilder.getSinkBuilder().setTypeClassName(typeArgs[1].getName()));
}
+
}
private Class<?> getTypeArg(String className, Class<?> funClass, URLClassLoader classLoader)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services