Fix: function load correct type-class of connector
@@ -21,6 +21,7 @@ package org.apache.pulsar.functions.worker;
import static org.apache.pulsar.functions.utils.Utils.FILE;
import static org.apache.pulsar.functions.utils.Utils.HTTP;
import static org.apache.pulsar.functions.utils.Utils.getSourceType;
+import static org.apache.pulsar.functions.utils.Utils.getSinkType;
import static org.apache.pulsar.functions.utils.Utils.isFunctionPackageUrlSupported;
import com.google.common.annotations.VisibleForTesting;
@@ -286,7 +287,7 @@ public class FunctionActioner implements AutoCloseable {
builder.setClassName(sourceClass);
functionDetails.setSource(builder);
- fillSourceSinkTypeClass(functionDetails, archive, sourceClass);
+ fillSourceTypeClass(functionDetails, archive, sourceClass);
return archive;
}
}
@@ -300,7 +301,7 @@ public class FunctionActioner implements AutoCloseable {
builder.setClassName(sinkClass);
functionDetails.setSink(builder);
- fillSourceSinkTypeClass(functionDetails, archive, sinkClass);
+ fillSinkTypeClass(functionDetails, archive, sinkClass);
return archive;
}
}
@@ -308,7 +309,7 @@ public class FunctionActioner implements AutoCloseable {
throw new IOException("Could not find built in archive definition");
}
- private void fillSourceSinkTypeClass(FunctionDetails.Builder functionDetails, File archive, String className)
+ private void fillSourceTypeClass(FunctionDetails.Builder functionDetails, File archive, String className)
throws IOException {
try (NarClassLoader ncl = NarClassLoader.getFromArchive(archive, Collections.emptySet())) {
String typeArg = getSourceType(className, ncl).getName();
@@ -316,6 +317,13 @@ public class FunctionActioner implements AutoCloseable {
SourceSpec.Builder sourceBuilder = SourceSpec.newBuilder(functionDetails.getSource());
sourceBuilder.setTypeClassName(typeArg);
functionDetails.setSource(sourceBuilder);
+ }
+ }
+
+ private void fillSinkTypeClass(FunctionDetails.Builder functionDetails, File archive, String className)
+ throws IOException {
+ try (NarClassLoader ncl = NarClassLoader.getFromArchive(archive, Collections.emptySet())) {
+ String typeArg = getSinkType(className, ncl).getName();
SinkSpec.Builder sinkBuilder = SinkSpec.newBuilder(functionDetails.getSink());
sinkBuilder.setTypeClassName(typeArg);
