{"record":{"id":"e6bb25b977e12ec1","repo":"apache/pulsar","slug":"sink-does-not-implement-correct-interface","errorCode":null,"errorMessage":"Sink does not implement correct interface","messagePattern":"Sink does not implement correct interface","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java","lineNumber":1146,"sourceCode":"                                conf.hasBatchingSpec() ? conf.getBatchingSpec() : null))\n                        .compressionType(FunctionCommon.convertFromFunctionDetailsCompressionType(\n                                conf.getCompressionType()));\n                pulsarSinkConfig.setProducerConfig(builder.build());\n\n                object = new PulsarSink(this.client, pulsarSinkConfig, this.properties, this.stats,\n                        this.functionClassLoader, this.producerCache);\n            }\n        } else {\n            object = Reflections.createInstance(\n                    sinkSpec.getClassName(),\n                    this.componentClassLoader);\n        }\n\n        if (object instanceof Sink) {\n            this.sink = (Sink) object;\n            this.sinkTypeArg = TypeResolver.resolveRawArguments(Sink.class, object.getClass())[0];\n        } else {\n            throw new RuntimeException(\"Sink does not implement correct interface\");\n        }\n\n        if (componentType == FunctionDetails.ComponentType.SINK) {\n            Thread.currentThread().setContextClassLoader(this.componentClassLoader);\n        }\n        try {\n            log.debug()\n                    .attr(\"sinkConfig\", sinkSpec.getConfigs())\n                    .attr(\"contextImpl\", contextImpl.toString())\n                    .log(\"Opening Sink\");\n            this.sink.open(augmentAndFilterConnectorConfig(sinkSpec.getConfigs()), contextImpl);\n        } catch (Exception e) {\n            log.error().exception(e).log(\"Sink open produced uncaught exception\");\n            throw e;\n        } finally {\n            Thread.currentThread().setContextClassLoader(this.instanceClassLoader);\n        }\n    }","sourceCodeStart":1128,"sourceCodeEnd":1164,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java#L1128-L1164","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Verify the configured className implements org.apache.pulsar.functions.api.Sink","Set pulsar-functions-api to provided scope so the interface is not duplicated inside the function JAR","Confirm the FunctionDetails componentType is SINK; redeploy with the correct component type","Clear the worker's functions extraction directory and restart workers to purge stale classloader state"],"exampleFix":"// before\nfunctions:\n  - className: com.acme.MyProcessor   # a Function, not a Sink\n    componentType: SINK\n// after\nfunctions:\n  - className: com.acme.MySink\n    componentType: SINK","handlingStrategy":"validation","validationCode":"Class<?> c = Class.forName(className);\nif (!Sink.class.isAssignableFrom(c)) {\n    throw new IllegalArgumentException(className + \" does not implement Sink\");\n}","typeGuard":"boolean isSink(Object o) { return o instanceof Sink; }","tryCatchPattern":"try { setupSink(); }\ncatch (RuntimeException e) {\n  if (e.getMessage().contains(\"Sink does not implement correct interface\")) {\n    log.error(\"Wrong class or duplicated Sink interface in classpath\", e);\n  }\n}","preventionTips":["Set pulsar-functions-api to provided scope in sink builds","Confirm componentType SINK matches a class implementing Sink","Purge worker function-extraction directories after JAR updates","Deploy sinks with an explicit, reviewed deployment manifest"],"tags":["java","pulsar-functions","classloader","sink","interface"],"backgroundTag":"wrong-interface-implementation","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}