{"record":{"id":"0cb7207b8f3b53c9","repo":"apache/pulsar","slug":"batchsourcetriggerer-does-not-implement-the-correc","errorCode":null,"errorMessage":"BatchSourceTriggerer does not implement the correct interface","messagePattern":"BatchSourceTriggerer 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":149,"sourceCode":"    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) {\n      discoveryTriggerer.init(batchSourceConfig.getDiscoveryTriggererConfig(),\n        this.sourceContext);\n      discoveryTriggerer.start(this::triggerDiscover);\n    }\n  }\n\n  volatile boolean discoverInProgress = false;\n  private synchronized void triggerDiscover(String discoveredEvent) {\n\n    if (discoverInProgress) {","sourceCodeStart":131,"sourceCodeEnd":167,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java#L131-L167","documentation":"In the same initializeBatchSource flow, the discovery triggerer class is instantiated via Reflections.createInstance and must implement BatchSourceTriggerer. Thrown when the configured discoveryTriggererClassName resolves to an object that does not implement that interface.","triggerScenarios":"batchSourceConfig.getDiscoveryTriggererClassName() names a class that instantiates successfully but is not a BatchSourceTriggerer, during open() -> initializeBatchSource.","commonSituations":"Wrong class name in the BatchSourceConfig (e.g. pointing at the BatchSource or an event-handler class); implementing the trigger logic against an older interface signature; copying a config between projects where the triggerer class changed.","solutions":["Set discoveryTriggererClassName to a class implementing org.apache.pulsar.io.core.BatchSourceTriggerer.","Implement init/hasMore/activate methods per the current BatchSourceTriggerer interface.","Rebuild the connector archive so the correct triggerer class is present in the NAR classloader."],"exampleFix":"// before\npublic class MyTriggerer implements SourceTriggerer { ... }\n// after\npublic class MyTriggerer implements BatchSourceTriggerer {\n  public void init(BatchSourceConfig cfg, Map<String,Object> conf, Consumer<byte[]> consumer) {...}\n  ...\n}","handlingStrategy":"validation","validationCode":"Class<?> c = Class.forName(discoveryTriggererClassName);\nif (!BatchSourceTriggerer.class.isAssignableFrom(c)) {\n  throw new IllegalArgumentException(discoveryTriggererClassName + \" is not a BatchSourceTriggerer\");\n}","typeGuard":"boolean isBatchSourceTriggerer(Object o) { return o instanceof BatchSourceTriggerer; }","tryCatchPattern":"try {\n  executor.open(config);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"BatchSourceTriggerer does not implement\")) {\n    log.error(\"discoveryTriggererClassName must implement BatchSourceTriggerer\");\n  }\n}","preventionTips":["Set discoveryTriggererClassName to a dedicated BatchSourceTriggerer implementation","Keep triggerer implementations against the current io.core interface","Include unit tests asserting the triggerer type before packaging the NAR"],"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"}