apache/pulsar · error · RuntimeException

Sink does not implement correct interface

Error message

Sink does not implement correct interface

What it means

Symmetric to the Source check: when setting up a Sink component, JavaInstanceRunnable verifies the loaded object implements org.apache.pulsar.functions.api.Sink and throws RuntimeException if not. The sink's type argument (sinkTypeArg) is resolved via TypeResolver only after this check passes.

Solutions

1. Verify the configured className implements org.apache.pulsar.functions.api.Sink
2. Set pulsar-functions-api to provided scope so the interface is not duplicated inside the function JAR
3. Confirm the FunctionDetails componentType is SINK; redeploy with the correct component type
4. Clear the worker's functions extraction directory and restart workers to purge stale classloader state

Example fix

// before
functions:
  - className: com.acme.MyProcessor   # a Function, not a Sink
    componentType: SINK
// after
functions:
  - className: com.acme.MySink
    componentType: SINK

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

boolean isSink(Object o) { return o instanceof Sink; }

try { setupSink(); }
catch (RuntimeException e) {
  if (e.getMessage().contains("Sink does not implement correct interface")) {
    log.error("Wrong class or duplicated Sink interface in classpath", e);
  }
}

When it happens

Trigger: A Function or Source class is configured as a Sink component; the Sink interface was bundled twice in the classpath so instanceof fails across classloaders; className points to the wrong class.

Common situations: Redeploying an existing function as a sink without changing className; uber JAR containing pulsar-functions-api classes; stale worker-side extracted JARs after a JAR update; copy-pasted deployment manifest with wrong component type.

Related errors

• Source does not implement correct interface
• Sink class does not implement the correct interface
• SourceRecord class type must be PulsarRecord
• Cannot find/load class
• Component cannot get state store

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

Appendix: source

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

                                conf.hasBatchingSpec() ? conf.getBatchingSpec() : null))
                        .compressionType(FunctionCommon.convertFromFunctionDetailsCompressionType(
                                conf.getCompressionType()));
                pulsarSinkConfig.setProducerConfig(builder.build());

                object = new PulsarSink(this.client, pulsarSinkConfig, this.properties, this.stats,
                        this.functionClassLoader, this.producerCache);
            }
        } else {
            object = Reflections.createInstance(
                    sinkSpec.getClassName(),
                    this.componentClassLoader);
        }

        if (object instanceof Sink) {
            this.sink = (Sink) object;
            this.sinkTypeArg = TypeResolver.resolveRawArguments(Sink.class, object.getClass())[0];
        } else {
            throw new RuntimeException("Sink does not implement correct interface");
        }

        if (componentType == FunctionDetails.ComponentType.SINK) {
            Thread.currentThread().setContextClassLoader(this.componentClassLoader);
        }
        try {
            log.debug()
                    .attr("sinkConfig", sinkSpec.getConfigs())
                    .attr("contextImpl", contextImpl.toString())
                    .log("Opening Sink");
            this.sink.open(augmentAndFilterConnectorConfig(sinkSpec.getConfigs()), contextImpl);
        } catch (Exception e) {
            log.error().exception(e).log("Sink open produced uncaught exception");
            throw e;
        } finally {
            Thread.currentThread().setContextClassLoader(this.instanceClassLoader);
        }
    }

View on GitHub (pinned to 820761864e)