{"record":{"id":"dcb427cfd3497924","repo":"apache/pulsar","slug":"source-class-s-does-not-implement-the-correct-int","errorCode":null,"errorMessage":"Source class %s does not implement the correct interface","messagePattern":"Source class (.+?) does not implement the correct interface","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java","lineNumber":212,"sourceCode":"        for (FunctionConfig.ProcessingGuarantees type : FunctionConfig.ProcessingGuarantees.values()) {\n            if (type.name().equals(processingGuarantees.name())) {\n                return type;\n            }\n        }\n        throw new RuntimeException(\"Unrecognized processing guarantee: \" + processingGuarantees.name());\n    }\n\n    public static TypeDefinition getSourceType(String className, TypePool typePool) {\n        return getSourceType(typePool.describe(className).resolve());\n    }\n\n    public static TypeDefinition getSourceType(TypeDefinition sourceClass) {\n        if (sourceClass.asErasure().isAssignableTo(Source.class)) {\n            return resolveInterfaceTypeArguments(sourceClass, Source.class).get(0);\n        } else if (sourceClass.asErasure().isAssignableTo(BatchSource.class)) {\n            return resolveInterfaceTypeArguments(sourceClass, BatchSource.class).get(0);\n        } else {\n            throw new IllegalArgumentException(\n              String.format(\"Source class %s does not implement the correct interface\",\n                sourceClass.getActualName()));\n        }\n    }\n\n    public static TypeDefinition getSinkType(String className, TypePool typePool) {\n        return getSinkType(typePool.describe(className).resolve());\n    }\n\n    public static TypeDefinition getSinkType(TypeDefinition sinkClass) {\n        if (sinkClass.asErasure().isAssignableTo(Sink.class)) {\n            return resolveInterfaceTypeArguments(sinkClass, Sink.class).get(0);\n        } else {\n            throw new IllegalArgumentException(\n                    String.format(\"Sink class %s does not implement the correct interface\",\n                            sinkClass.getActualName()));\n        }\n    }","sourceCodeStart":194,"sourceCodeEnd":230,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java#L194-L230","documentation":"FunctionCommon.getSourceType inspects a user-provided Source class via bytecode analysis to determine its output type. It only accepts classes assignable to org.apache.pulsar.functions.api.Source or BatchSource; otherwise it throws this IllegalArgumentException naming the offending class.","triggerScenarios":"Passing a class to getSourceType(String className, TypePool) / getSourceType(TypeDefinition) that does not implement Source or BatchSource — e.g. a PulsarFunction/Sink class supplied as the source class of a function or connector config, a class implementing a removed/renamed old source interface, or a typo'd className resolving to the wrong class.","commonSituations":"Configuring a connector with the wrong main class in a NAR/yard archive; migrating from the deprecated PulsarSource/PulsarIO interfaces to the java-function api; typos in functionConfig.getClassName; source JAR missing a compile-time dependency making the class fail assignability resolution.","solutions":["Make the class implement org.apache.pulsar.functions.api.Source<T> (or BatchSource<T>) with a concrete type argument","Verify the configured class name points at the intended source class, not a sink or plain function class","If migrating from the old pulsar-io API, re-implement the connector against the org.apache.pulsar.functions.api Source interface","Check the packaged jar actually contains the compiled class implementing the correct interface"],"exampleFix":"// before\npublic class MyConnector implements Sink<String> { ... } // used as source\n// after\npublic class MyConnector implements Source<String> {\n  @Override public void open(Map<String,Object> config, SourceContext ctx) { }\n  @Override public Record<String> read() { return null; }\n  @Override public void close() { }\n}","handlingStrategy":"validation","validationCode":"// Verify interface before calling getSourceType\nClass<?> clazz = Class.forName(className);\nif (!org.apache.pulsar.functions.api.Source.class.isAssignableFrom(clazz)\n    && !org.apache.pulsar.functions.api.BatchSource.class.isAssignableFrom(clazz)) {\n  throw new IllegalArgumentException(className + \" must implement Source or BatchSource\");\n}","typeGuard":"static boolean isValidSourceClass(String className) throws ClassNotFoundException {\n  Class<?> c = Class.forName(className);\n  return org.apache.pulsar.functions.api.Source.class.isAssignableFrom(c)\n      || org.apache.pulsar.functions.api.BatchSource.class.isAssignableFrom(c);\n}","tryCatchPattern":"try {\n  TypeDefinition td = FunctionCommon.getSourceType(className, typePool);\n} catch (IllegalArgumentException e) {\n  log.error(\"Bad source class: {}\", e.getMessage());\n  throw new InvalidFunctionDefinitionException(className + \" is not a Source/BatchSource\");\n}","preventionTips":["Always implement org.apache.pulsar.functions.api.Source<T> (or BatchSource<T>) with a concrete type argument","Double-check the className configured in sourceConfig before submission","When migrating legacy connectors, port to the java-function API interfaces","Smoke-test connector archives locally before uploading"],"tags":["pulsar-functions","source","interface","configuration"],"backgroundTag":"class-does-not-implement-interface","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}