{"record":{"id":"7af22c4a06d42d03","repo":"apache/pulsar","slug":"source-does-not-implement-correct-interface","errorCode":null,"errorMessage":"Source does not implement correct interface","messagePattern":"Source does not implement correct interface","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java","lineNumber":962,"sourceCode":"\n            // check if source is a batch source\n            if (sourceSpec.getClassName().equals(BatchSourceExecutor.class.getName())) {\n                object = Reflections.createInstance(\n                  sourceSpec.getClassName(),\n                  this.instanceClassLoader);\n            } else {\n                object = Reflections.createInstance(\n                  sourceSpec.getClassName(),\n                  this.componentClassLoader);\n            }\n        }\n\n        Class<?>[] typeArgs;\n        if (object instanceof Source) {\n            typeArgs = TypeResolver.resolveRawArguments(Source.class, object.getClass());\n            assert typeArgs.length > 0;\n        } else {\n            throw new RuntimeException(\"Source does not implement correct interface\");\n        }\n        this.source = (Source<?>) object;\n\n        if (componentType == FunctionDetails.ComponentType.SOURCE) {\n            Thread.currentThread().setContextClassLoader(this.componentClassLoader);\n        }\n        try {\n            this.source.open(augmentAndFilterConnectorConfig(sourceSpec.getConfigs()), contextImpl);\n            if (this.source instanceof PulsarSource) {\n                contextImpl.setInputConsumers(((PulsarSource) this.source).getInputConsumers());\n            }\n        } catch (Exception e) {\n            log.error().exception(e).log(\"Source open produced uncaught exception\");\n            throw e;\n        } finally {\n            Thread.currentThread().setContextClassLoader(this.instanceClassLoader);\n        }\n    }","sourceCodeStart":944,"sourceCodeEnd":980,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java#L944-L980","documentation":"During instance setup, setupInput loads the user's Source class and verifies it is an instance of org.apache.pulsar.functions.api.Source before resolving type arguments. If the loaded object does not implement the Source interface, the instance cannot wire the record pipeline and throws RuntimeException. This is almost always a class-loading or packaging problem, not a coding one.","triggerScenarios":"The configured class name resolves (Class.forName succeeds) but the loaded Class was loaded by a different classloader than the one that loaded the Source interface, so instanceof fails; or the user supplied a Function/Sink class in a Source component's configuration.","commonSituations":"Function JAR bundled inside the fat/uber JAR duplicating pulsar-functions-api classes so two Source interface classes exist; wrong component type selected at deployment (source vs function vs sink); className typo pointing at the wrong class; stale JAR in functions worker's nar/extraction directory.","solutions":["Verify the configured className actually implements org.apache.pulsar.functions.api.Source","Mark the pulsar-functions-api dependency as provided in the build so API classes are NOT bundled into the function JAR (two copies of the interface break instanceof)","Confirm the component type in FunctionDetails is SOURCE for a source class; redeploy with the correct type","Delete stale extracted function JARs on the worker (pulsar_functions extraction dir) and restart workers so the fresh JAR is used"],"exampleFix":"// before (pom.xml)\n<dependency>\n  <groupId>org.apache.pulsar</groupId>\n  <artifactId>pulsar-functions-api</artifactId>\n  <version>3.x</version>\n</dependency>\n// after\n<dependency>\n  <groupId>org.apache.pulsar</groupId>\n  <artifactId>pulsar-functions-api</artifactId>\n  <version>3.x</version>\n  <scope>provided</scope>\n</dependency>","handlingStrategy":"validation","validationCode":"Class<?> c = Class.forName(className);\nif (!Source.class.isAssignableFrom(c)) {\n    throw new IllegalArgumentException(className + \" does not implement Source\");\n}","typeGuard":"boolean isSource(Object o) { return o instanceof Source; }","tryCatchPattern":"try { setupSource(); }\ncatch (RuntimeException e) {\n  if (e.getMessage().contains(\"does not implement correct interface\")) {\n    log.error(\"Check componentType and ensure pulsar-functions-api is not bundled in the JAR\", e);\n  }\n}","preventionTips":["Mark pulsar-functions-api as provided scope; never bundle it in the function JAR","Verify className implements Source and componentType is SOURCE before deploying","Clear stale extracted function JARs on workers after updates","Use the same Pulsar version for the API dependency as the cluster"],"tags":["java","pulsar-functions","classloader","source","interface"],"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-14T00:17:10.932Z"}