apache/pulsar · error · IllegalArgumentException

Source class %s does not implement the correct interface

Error message

Source class %s does not implement the correct interface

What it means

FunctionCommon.getSourceType inspects a user-provided Source class via bytecode analysis to determine its output type. It only accepts classes assignable to org.apache.pulsar.functions.api.Source or BatchSource; otherwise it throws this IllegalArgumentException naming the offending class.

Source

Thrown at pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java:212

        for (FunctionConfig.ProcessingGuarantees type : FunctionConfig.ProcessingGuarantees.values()) {
            if (type.name().equals(processingGuarantees.name())) {
                return type;
            }
        }
        throw new RuntimeException("Unrecognized processing guarantee: " + processingGuarantees.name());
    }

    public static TypeDefinition getSourceType(String className, TypePool typePool) {
        return getSourceType(typePool.describe(className).resolve());
    }

    public static TypeDefinition getSourceType(TypeDefinition sourceClass) {
        if (sourceClass.asErasure().isAssignableTo(Source.class)) {
            return resolveInterfaceTypeArguments(sourceClass, Source.class).get(0);
        } else if (sourceClass.asErasure().isAssignableTo(BatchSource.class)) {
            return resolveInterfaceTypeArguments(sourceClass, BatchSource.class).get(0);
        } else {
            throw new IllegalArgumentException(
              String.format("Source class %s does not implement the correct interface",
                sourceClass.getActualName()));
        }
    }

    public static TypeDefinition getSinkType(String className, TypePool typePool) {
        return getSinkType(typePool.describe(className).resolve());
    }

    public static TypeDefinition getSinkType(TypeDefinition sinkClass) {
        if (sinkClass.asErasure().isAssignableTo(Sink.class)) {
            return resolveInterfaceTypeArguments(sinkClass, Sink.class).get(0);
        } else {
            throw new IllegalArgumentException(
                    String.format("Sink class %s does not implement the correct interface",
                            sinkClass.getActualName()));
        }
    }

View on GitHub (pinned to 820761864e)

Solutions

1. Make the class implement org.apache.pulsar.functions.api.Source<T> (or BatchSource<T>) with a concrete type argument
2. Verify the configured class name points at the intended source class, not a sink or plain function class
3. If migrating from the old pulsar-io API, re-implement the connector against the org.apache.pulsar.functions.api Source interface
4. Check the packaged jar actually contains the compiled class implementing the correct interface

Example fix

// before
public class MyConnector implements Sink<String> { ... } // used as source
// after
public class MyConnector implements Source<String> {
  @Override public void open(Map<String,Object> config, SourceContext ctx) { }
  @Override public Record<String> read() { return null; }
  @Override public void close() { }
}

// Verify interface before calling getSourceType
Class<?> clazz = Class.forName(className);
if (!org.apache.pulsar.functions.api.Source.class.isAssignableFrom(clazz)
    && !org.apache.pulsar.functions.api.BatchSource.class.isAssignableFrom(clazz)) {
  throw new IllegalArgumentException(className + " must implement Source or BatchSource");
}

static boolean isValidSourceClass(String className) throws ClassNotFoundException {
  Class<?> c = Class.forName(className);
  return org.apache.pulsar.functions.api.Source.class.isAssignableFrom(c)
      || org.apache.pulsar.functions.api.BatchSource.class.isAssignableFrom(c);
}

try {
  TypeDefinition td = FunctionCommon.getSourceType(className, typePool);
} catch (IllegalArgumentException e) {
  log.error("Bad source class: {}", e.getMessage());
  throw new InvalidFunctionDefinitionException(className + " is not a Source/BatchSource");
}

When it happens

Trigger: Passing a class to getSourceType(String className, TypePool) / getSourceType(TypeDefinition) that does not implement Source or BatchSource — e.g. a PulsarFunction/Sink class supplied as the source class of a function or connector config, a class implementing a removed/renamed old source interface, or a typo'd className resolving to the wrong class.

Common situations: Configuring a connector with the wrong main class in a NAR/yard archive; migrating from the deprecated PulsarSource/PulsarIO interfaces to the java-function api; typos in functionConfig.getClassName; source JAR missing a compile-time dependency making the class fail assignability resolution.

Related errors

• Source does not implement correct interface
• Sink class %s does not implement the correct interface
• Unknown component type: %s
• Access to environment variable %s is not allowed.
• PulsarAdmin is not enabled in function worker

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