{"record":{"id":"0e976b212405b2d2","repo":"apache/pulsar","slug":"batchsource-does-not-implement-the-correct-interfa","errorCode":null,"errorMessage":"BatchSource does not implement the correct interface","messagePattern":"BatchSource does not implement the correct interface","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java","lineNumber":139,"sourceCode":"    }\n\n    String batchSourceConfigJson = (String) config.get(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY);\n    this.batchSourceConfig = new Gson().fromJson(batchSourceConfigJson, BatchSourceConfig.class);\n    this.batchSourceClassName = (String) config.get(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY);\n  }\n\n  private void initializeBatchSource() {\n    // First init the batchsource\n    ClassLoader clsLoader = Thread.currentThread().getContextClassLoader();\n    Object userClassObject = Reflections.createInstance(\n      batchSourceClassName,\n      clsLoader);\n    if (userClassObject instanceof BatchSource) {\n      @SuppressWarnings(\"unchecked\") // type parameter is erased at runtime\n      BatchSource<T> typedBatchSource = (BatchSource<T>) userClassObject;\n      batchSource = typedBatchSource;\n    } else {\n      throw new IllegalArgumentException(\"BatchSource does not implement the correct interface\");\n    }\n\n    // next init the discovery triggerer\n    Object discoveryClassObject = Reflections.createInstance(\n      batchSourceConfig.getDiscoveryTriggererClassName(),\n      clsLoader);\n    if (discoveryClassObject instanceof BatchSourceTriggerer) {\n      discoveryTriggerer = (BatchSourceTriggerer) discoveryClassObject;\n    } else {\n      throw new IllegalArgumentException(\"BatchSourceTriggerer does not implement the correct interface\");\n    }\n  }\n\n  private void start() throws Exception {\n    isRunning = true;\n    createIntermediateTopicConsumer();\n    batchSource.open(this.config, this.sourceContext);\n    if (sourceContext.getInstanceId() == 0) {","sourceCodeStart":121,"sourceCodeEnd":157,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java#L121-L157","documentation":"BatchSourceExecutor.initializeBatchSource instantiates the configured class via reflection and requires it to implement org.apache.pulsar.io.core.BatchSource. Thrown when the class at batchSourceClassName loads and instantiates but is not a BatchSource.","triggerScenarios":"The classname given for the batch source points to a class implementing Source/other interface (or no interface), passed through open() -> initializeBatchSource during source startup.","commonSituations":"Typo/wrong class in the submission (pointing at the RecordReader or triggerer class instead of the BatchSource); wrong archive submitted so the classloader resolves a different class; refactoring dropped the BatchSource interface.","solutions":["Ensure the configured class implements org.apache.pulsar.io.core.BatchSource (extends BatchSource implements RecordReader usage).","Correct the classname in the batch source submission config to the BatchSource implementation, not a helper class.","Rebuild the NAR/archive and verify it contains the correct class via jar tf."],"exampleFix":"// before\npublic class MySource implements Source { ... }\n// after\npublic class MySource extends BatchSource<byte[]> { ... }","handlingStrategy":"validation","validationCode":"Class<?> c = Class.forName(batchSourceClassName);\nif (!BatchSource.class.isAssignableFrom(c)) {\n  throw new IllegalArgumentException(batchSourceClassName + \" is not a BatchSource\");\n}","typeGuard":"boolean isBatchSource(Object o) { return o instanceof BatchSource; }","tryCatchPattern":"try {\n  executor.open(config);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"BatchSource does not implement\")) {\n    log.error(\"configured class is not a BatchSource; check submission classname\");\n  }\n}","preventionTips":["Point the submission at the class extending BatchSource, not helpers","Verify class presence in the NAR archive with jar tf","Re-run interface checks after refactoring connector code"],"tags":["batch-source","reflection","class-cast","pulsar-functions"],"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"}