apache/pulsar · error · IOException

The '%s' connector does not provide a sink implementation

Error message

The '%s' connector does not provide a sink implementation

What it means

ConnectorUtils.getIOSinkClass extracts the Pulsar IO Sink class from a connector NAR. If the connector definition's sinkClass field is empty, the archive declares no sink implementation and an IOException is thrown.

Source

Thrown at pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java:95

            Class<?> sourceClass = narClassLoader.loadClass(conf.getSourceClass());
            if (!(Source.class.isAssignableFrom(sourceClass) || BatchSource.class.isAssignableFrom(sourceClass))) {
                throw new IOException(String.format("Class %s does not implement interface %s or %s",
                        conf.getSourceClass(), Source.class.getName(), BatchSource.class.getName()));
            }
        } catch (Throwable t) {
            Exceptions.rethrowIOException(t);
        }

        return conf.getSourceClass();
    }

    /**
     * Extract the Pulsar IO Sink class from a connector archive.
     */
    public static String getIOSinkClass(NarClassLoader narClassLoader) throws IOException {
        ConnectorDefinition conf = getConnectorDefinition(narClassLoader);
        if (StringUtils.isEmpty(conf.getSinkClass())) {
            throw new IOException(
                    String.format("The '%s' connector does not provide a sink implementation", conf.getName()));
        }

        try {
            // Try to load sink class and check it implements Sink interface
            Class<?> sinkClass = narClassLoader.loadClass(conf.getSinkClass());
            if (!(Sink.class.isAssignableFrom(sinkClass))) {
                throw new IOException(
                        "Class " + conf.getSinkClass() + " does not implement interface " + Sink.class.getName());
            }
        } catch (Throwable t) {
            Exceptions.rethrowIOException(t);
        }

        return conf.getSinkClass();
    }

    public static ConnectorDefinition getConnectorDefinition(File narFile) throws IOException {

View on GitHub (pinned to 820761864e)

Solutions

Use a connector NAR that implements a Sink
Add the sinkClass entry to the connector definition and rebuild the NAR
Inspect the definition via getConnectorDefinition() and confirm getSinkClass() is set before calling

Example fix

// before
String cls = ConnectorUtils.getIOSinkClass(sourceOnlyLoader); // throws
// after
ConnectorDefinition def = ConnectorUtils.getConnectorDefinition(loader);
if (StringUtils.isNotEmpty(def.getSinkClass())) {
    String cls = ConnectorUtils.getIOSinkClass(loader);
}

Defensive patterns

Strategy: validation

Validate before calling

ConnectorDefinition def = ConnectorUtils.getConnectorDefinition(narClassLoader);
if (def == null || StringUtils.isEmpty(def.getSinkClass())) {
    throw new IllegalArgumentException("NAR is not a sink connector: " + def.getName());
}

Type guard

boolean isSinkConnector(ConnectorDefinition def) {
    return def != null && StringUtils.isNotEmpty(def.getSinkClass());
}

Try / catch

try {
    String sink = ConnectorUtils.getIOSinkClass(loader);
} catch (IOException e) {
    if (e.getMessage().contains("does not provide a sink implementation")) {
        // treat as source-only connector
    } else { throw e; }
}

Prevention

Check connector docs for source/sink support before deploying as a sink
Validate NARs at upload time with getConnectorDefinition
Avoid mixing source and sink archives in the connectors directory

When it happens

Trigger: Calling getIOSinkClass(narClassLoader) on a source-only connector NAR (definition has no sinkClass), e.g. when creating a sink with a source connector archive.

Common situations: Deploying a source-only connector (e.g. Kafka source NAR) as a sink; missing sinkClass field in connector metadata; wrong NAR uploaded to the functions worker's connector directory.

Related errors

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