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 842da22f176d5301fba5ef6d8481761e198b67dd..43e034682a57f6bb248856ec7db460dfd702b78d 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 @@ -369,6 +369,20 @@ public class FunctionConfigUtils { private static void doJavaChecks(FunctionConfig functionConfig, ClassLoader clsLoader) { + try { + Class functionClass = clsLoader.loadClass(functionConfig.getClassName()); + + if (!org.apache.pulsar.functions.api.Function.class.isAssignableFrom(functionClass) + && !java.util.function.Function.class.isAssignableFrom(functionClass)) { + throw new IllegalArgumentException( + String.format("Function class %s does not implement the correct interface", + functionClass.getName())); + } + } catch (ClassNotFoundException e) { + throw new IllegalArgumentException( + String.format("Function class %s must be in class path", functionConfig.getClassName()), e); + } + Class[] typeArgs; try { typeArgs = FunctionCommon.getFunctionTypes(functionConfig, clsLoader); diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3ResourceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3ResourceTest.java index e5df566a5a18e6ad90f39ce1db71162e6dc42133..5f90945983e7fa2c436b9ac60f5da70567770df7 100644 --- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3ResourceTest.java +++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3ResourceTest.java @@ -43,6 +43,7 @@ import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.function.Consumer; import javax.ws.rs.core.Response; import javax.ws.rs.core.StreamingOutput; @@ -109,6 +110,13 @@ public class FunctionApiV3ResourceTest { } } + private static final class WrongFunction implements Consumer { + @Override + public void accept(String s) { + + } + } + private static final String tenant = "test-tenant"; private static final String namespace = "test-namespace"; private static final String function = "test-function"; @@ -438,6 +446,27 @@ public class FunctionApiV3ResourceTest { } } + @Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "Function class .*. does not implement the correct interface") + public void testRegisterFunctionImplementWrongInterface() { + try { + testRegisterFunctionMissingArguments( + tenant, + namespace, + function, + mockedInputStream, + topicsToSerDeClassName, + mockedFormData, + outputTopic, + outputSerdeClassName, + WrongFunction.class.getName(), + parallelism, + null); + } catch (RestException re){ + assertEquals(re.getResponse().getStatusInfo(), Response.Status.BAD_REQUEST); + throw re; + } + } + private void testRegisterFunctionMissingArguments( String tenant, String namespace,