{"record":{"id":"7a110386c9c97b28","repo":"apache/pulsar","slug":"source-class-s-does-not-implement-the-correct-int-7a1103","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/SourceConfigUtils.java","lineNumber":316,"sourceCode":"            }\n            sourceClassName = connectorDefinition.getSourceClass();\n            if (sourceClassName == null) {\n                throw new IllegalArgumentException(\"Failed to extract source class from archive\");\n            }\n        }\n\n        // check if source implements the correct interfaces\n        TypeDescription sourceClass;\n        try {\n            sourceClass = sourceFunction.resolveType(sourceClassName);\n        } catch (TypePool.Resolution.NoSuchTypeException e) {\n            throw new IllegalArgumentException(\n              String.format(\"Source class %s not found in class loader\", sourceClassName), e);\n        }\n\n        if (!(sourceClass.asErasure().isAssignableTo(Source.class) || sourceClass.asErasure()\n                .isAssignableTo(BatchSource.class))) {\n            throw new IllegalArgumentException(\n                    String.format(\"Source class %s does not implement the correct interface\",\n                            sourceClass.getName()));\n        }\n\n        if (sourceClass.asErasure().isAssignableTo(BatchSource.class)) {\n            if (sourceConfig.getBatchSourceConfig() != null) {\n                validateBatchSourceConfig(sourceConfig.getBatchSourceConfig());\n            } else {\n                throw new IllegalArgumentException(\n                  String.format(\"Source class %s implements %s but batch source source config is not specified\",\n                    sourceClass.getName(), BatchSource.class.getName()));\n            }\n        }\n\n        // extract type from source class\n        TypeDefinition typeArg;\n\n        try {","sourceCodeStart":298,"sourceCodeEnd":334,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java#L298-L334","documentation":"validateAndExtractDetails verifies that the resolved source class implements org.apache.pulsar.io.core.Source or BatchSource via ByteBuddy erasure assignability checks. If it implements neither, the class loads fine but is not a usable source, so this IllegalArgumentException is thrown with the class name.","triggerScenarios":"Setting className (or a pulsar-io.yaml sourceClass) to a class that doesn't implement Source/BatchSource — e.g. pointing at a Sink implementation, a helper class, or a custom connector class missing the 'implements Source' declaration; also BatchSource implementations that require BatchSourceConfig handling.","commonSituations":"Registering a sink connector through the sources API; copying a className from a sink config into a source config; writing a connector class that implements the wrong interface or an old internal one; generically validating a package against the wrong function type.","solutions":["Point className/sourceClass at a class that implements org.apache.pulsar.io.core.Source (or BatchSource)","If the class is a Sink, register it via pulsar-admin sinks instead of sources","If it's your own connector, declare 'implements Source<T>' and implement open/recordRead etc.","Confirm with 'implements BatchSource<T>' plus a BatchSourceConfig when doing batch ingestion, and set batchSourceConfig accordingly"],"exampleFix":"// before\npublic class MyConnector implements Sink<String> { ... } // registered as source\n// after\npublic class MyConnector implements Source<String> {\n    public void open(Map<String, Object> config, SourceContext ctx) { ... }\n    public Record<String> read() throws Exception { ... }\n}","handlingStrategy":"validation","validationCode":"// Confirm the class implements the right interface before submit:\nClass<?> c = Class.forName(cfg.getClassName(), false, getClass().getClassLoader());\nif (!org.apache.pulsar.io.core.Source.class.isAssignableFrom(c)\n        && !org.apache.pulsar.io.core.BatchSource.class.isAssignableFrom(c)) {\n    throw new IllegalStateException(cfg.getClassName() + \" does not implement Source/BatchSource\");\n}","typeGuard":"static boolean isSourceClass(Class<?> c) {\n    return org.apache.pulsar.io.core.Source.class.isAssignableFrom(c)\n        || org.apache.pulsar.io.core.BatchSource.class.isAssignableFrom(c);\n}","tryCatchPattern":"try {\n    SourceConfigUtils.validateAndExtractDetails(cfg, pkg, true);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"does not implement the correct interface\")) {\n        // wrong connector type: use the sinks API or pick a class implementing Source\n    }\n}","preventionTips":["Only submit classes implementing org.apache.pulsar.io.core.Source or BatchSource to the sources API","Register Sink implementations through the sinks API, not sources","Declare 'implements Source<T>' (with open/read) when writing custom connectors","For batch ingestion use BatchSource plus batchSourceConfig in the SourceConfig"],"tags":["pulsar-functions","connector-interface","type-mismatch","pulsar-io"],"backgroundTag":"wrong-interface-implementation","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}