apache/pulsar · error · RuntimeException

Source does not implement correct interface

Error message

Source does not implement correct interface

What it means

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.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java:962

            // check if source is a batch source
            if (sourceSpec.getClassName().equals(BatchSourceExecutor.class.getName())) {
                object = Reflections.createInstance(
                  sourceSpec.getClassName(),
                  this.instanceClassLoader);
            } else {
                object = Reflections.createInstance(
                  sourceSpec.getClassName(),
                  this.componentClassLoader);
            }
        }

        Class<?>[] typeArgs;
        if (object instanceof Source) {
            typeArgs = TypeResolver.resolveRawArguments(Source.class, object.getClass());
            assert typeArgs.length > 0;
        } else {
            throw new RuntimeException("Source does not implement correct interface");
        }
        this.source = (Source<?>) object;

        if (componentType == FunctionDetails.ComponentType.SOURCE) {
            Thread.currentThread().setContextClassLoader(this.componentClassLoader);
        }
        try {
            this.source.open(augmentAndFilterConnectorConfig(sourceSpec.getConfigs()), contextImpl);
            if (this.source instanceof PulsarSource) {
                contextImpl.setInputConsumers(((PulsarSource) this.source).getInputConsumers());
            }
        } catch (Exception e) {
            log.error().exception(e).log("Source open produced uncaught exception");
            throw e;
        } finally {
            Thread.currentThread().setContextClassLoader(this.instanceClassLoader);
        }
    }

View on GitHub (pinned to 820761864e)

Solutions

1. Verify the configured className actually implements org.apache.pulsar.functions.api.Source
2. 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)
3. Confirm the component type in FunctionDetails is SOURCE for a source class; redeploy with the correct type
4. Delete stale extracted function JARs on the worker (pulsar_functions extraction dir) and restart workers so the fresh JAR is used

Example fix

// before (pom.xml)
<dependency>
  <groupId>org.apache.pulsar</groupId>
  <artifactId>pulsar-functions-api</artifactId>
  <version>3.x</version>
</dependency>
// after
<dependency>
  <groupId>org.apache.pulsar</groupId>
  <artifactId>pulsar-functions-api</artifactId>
  <version>3.x</version>
  <scope>provided</scope>
</dependency>

Class<?> c = Class.forName(className);
if (!Source.class.isAssignableFrom(c)) {
    throw new IllegalArgumentException(className + " does not implement Source");
}

boolean isSource(Object o) { return o instanceof Source; }

try { setupSource(); }
catch (RuntimeException e) {
  if (e.getMessage().contains("does not implement correct interface")) {
    log.error("Check componentType and ensure pulsar-functions-api is not bundled in the JAR", e);
  }
}

When it happens

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

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

Related errors

• Sink does not implement correct interface
• The record returned by the source cannot be null
• Source class %s does not implement the correct interface
• Field '${name}' with value '${o}' does not implement ${class
• Cannot find/load class ${className}

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