[GitHub] merlimat closed pull request #2147: Fix: function load correct type-class of connector

[GitBox <gi...@apache.org>]

merlimat closed pull request #2147: Fix: function load correct type-class of connector
URL: https://github.com/apache/incubator-pulsar/pull/2147
 
 
   

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-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 4741c404d6..8ad292110d 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
@@ -21,6 +21,7 @@
 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 @@ private File getBuiltinArchive(FunctionDetails.Builder functionDetails) throws I
                 builder.setClassName(sourceClass);
                 functionDetails.setSource(builder);
 
-                fillSourceSinkTypeClass(functionDetails, archive, sourceClass);
+                fillSourceTypeClass(functionDetails, archive, sourceClass);
                 return archive;
             }
         }
@@ -300,7 +301,7 @@ private File getBuiltinArchive(FunctionDetails.Builder functionDetails) throws I
                 builder.setClassName(sinkClass);
                 functionDetails.setSink(builder);
 
-                fillSourceSinkTypeClass(functionDetails, archive, sinkClass);
+                fillSinkTypeClass(functionDetails, archive, sinkClass);
                 return archive;
             }
         }
@@ -308,7 +309,7 @@ private File getBuiltinArchive(FunctionDetails.Builder functionDetails) throws I
         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 @@ private void fillSourceSinkTypeClass(FunctionDetails.Builder functionDetails, Fi
             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);


 

----------------------------------------------------------------
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