apache/pulsar · error · IllegalArgumentException

BatchSource does not implement the correct interface

Error message

BatchSource does not implement the correct interface

What it means

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.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java:139

    }

    String batchSourceConfigJson = (String) config.get(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY);
    this.batchSourceConfig = new Gson().fromJson(batchSourceConfigJson, BatchSourceConfig.class);
    this.batchSourceClassName = (String) config.get(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY);
  }

  private void initializeBatchSource() {
    // First init the batchsource
    ClassLoader clsLoader = Thread.currentThread().getContextClassLoader();
    Object userClassObject = Reflections.createInstance(
      batchSourceClassName,
      clsLoader);
    if (userClassObject instanceof BatchSource) {
      @SuppressWarnings("unchecked") // type parameter is erased at runtime
      BatchSource<T> typedBatchSource = (BatchSource<T>) userClassObject;
      batchSource = typedBatchSource;
    } else {
      throw new IllegalArgumentException("BatchSource does not implement the correct interface");
    }

    // next init the discovery triggerer
    Object discoveryClassObject = Reflections.createInstance(
      batchSourceConfig.getDiscoveryTriggererClassName(),
      clsLoader);
    if (discoveryClassObject instanceof BatchSourceTriggerer) {
      discoveryTriggerer = (BatchSourceTriggerer) discoveryClassObject;
    } else {
      throw new IllegalArgumentException("BatchSourceTriggerer does not implement the correct interface");
    }
  }

  private void start() throws Exception {
    isRunning = true;
    createIntermediateTopicConsumer();
    batchSource.open(this.config, this.sourceContext);
    if (sourceContext.getInstanceId() == 0) {

View on GitHub (pinned to 820761864e)

Solutions

1. Ensure the configured class implements org.apache.pulsar.io.core.BatchSource (extends BatchSource implements RecordReader usage).
2. Correct the classname in the batch source submission config to the BatchSource implementation, not a helper class.
3. Rebuild the NAR/archive and verify it contains the correct class via jar tf.

Example fix

// before
public class MySource implements Source { ... }
// after
public class MySource extends BatchSource<byte[]> { ... }

Class<?> c = Class.forName(batchSourceClassName);
if (!BatchSource.class.isAssignableFrom(c)) {
  throw new IllegalArgumentException(batchSourceClassName + " is not a BatchSource");
}

boolean isBatchSource(Object o) { return o instanceof BatchSource; }

try {
  executor.open(config);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("BatchSource does not implement")) {
    log.error("configured class is not a BatchSource; check submission classname");
  }
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• BatchSourceTriggerer does not implement the correct interfac
• functions may not take more than two arguments, but function
• function takes two arguments, but the first is not Context.
• function may not return more than two values
• function returns two values, but the second does not impleme

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/0e976b212405b2d2. Report an issue: GitHub.