apache/pulsar · error · IllegalArgumentException

Sink class %s does not implement the correct interface

Error message

Sink class %s does not implement the correct interface

What it means

FunctionCommon.getSinkType inspects a user-provided Sink class to derive its input type via bytecode analysis. It requires the class to be assignable to org.apache.pulsar.functions.api.Sink; otherwise this IllegalArgumentException naming the offending class is thrown.

Source

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

            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()));
        }
    }

    public static void downloadFromHttpUrl(String destPkgUrl, File targetFile) throws IOException {
        final URL url = new URL(destPkgUrl);
        final URLConnection connection = url.openConnection();
        if (StringUtils.isNotEmpty(url.getUserInfo())) {
            final AuthenticationDataBasic authBasic = new AuthenticationDataBasic(url.getUserInfo());
            for (Map.Entry<String, String> header : authBasic.getHttpHeaders()) {
                connection.setRequestProperty(header.getKey(), header.getValue());
            }
        }
        try (InputStream in = connection.getInputStream()) {
            log.info().attr("url", destPkgUrl).attr("target", targetFile.getAbsoluteFile())
                    .log("Downloading function package");
            Files.copy(in, targetFile.toPath(), StandardCopyOption.REPLACE_EXISTING);

View on GitHub (pinned to 820761864e)

Solutions

1. Make the class implement org.apache.pulsar.functions.api.Sink<T> with a concrete type argument
2. Correct the sinkConfig className so it references the actual Sink implementation
3. If migrating from the legacy io API, port the connector to org.apache.pulsar.functions.api.Sink
4. Verify the jar on the package URL contains the class with an unshaded/unrelocated Sink supertype

Example fix

// before
public class MySink implements Source<String> { ... } // registered as sink
// after
public class MySink implements Sink<String> {
  @Override public void open(Map<String,Object> config, SinkContext ctx) { }
  @Override public void write(Record<String> record) { }
  @Override public void close() { }
}

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

static boolean isValidSinkClass(String className) throws ClassNotFoundException {
  return org.apache.pulsar.functions.api.Sink.class.isAssignableFrom(Class.forName(className));
}

try {
  TypeDefinition td = FunctionCommon.getSinkType(className, typePool);
} catch (IllegalArgumentException e) {
  log.error("Bad sink class: {}", e.getMessage());
  throw new InvalidFunctionDefinitionException(className + " is not a Sink");
}

When it happens

Trigger: Calling getSinkType(String className, TypePool) / getSinkType(TypeDefinition) with a class that does not implement Sink — e.g. passing a Source implementation as the sink class, a class implementing an old/renamed sink interface (pre java-function API), or a misconfigured className.

Common situations: Sink connector NAR built with the wrong entry class; migrating legacy pulsar-io sinks to the org.apache.pulsar.functions.api.Sink interface; typo in the sink class name in sinkConfig; shaded/relocated Sink interface in a fat jar breaking assignability.

Related errors

• Sink does not implement correct interface
• Source class %s does not implement the correct interface
• Sink class %s not found
• Transform function class name must be set
• Function class %s not found

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