{"record":{"id":"f77bd4e83a54f742","repo":"apache/pulsar","slug":"the-s-connector-does-not-provide-a-sink-impleme","errorCode":null,"errorMessage":"The '%s' connector does not provide a sink implementation","messagePattern":"The '(.+?)' connector does not provide a sink implementation","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java","lineNumber":95,"sourceCode":"            Class<?> sourceClass = narClassLoader.loadClass(conf.getSourceClass());\n            if (!(Source.class.isAssignableFrom(sourceClass) || BatchSource.class.isAssignableFrom(sourceClass))) {\n                throw new IOException(String.format(\"Class %s does not implement interface %s or %s\",\n                        conf.getSourceClass(), Source.class.getName(), BatchSource.class.getName()));\n            }\n        } catch (Throwable t) {\n            Exceptions.rethrowIOException(t);\n        }\n\n        return conf.getSourceClass();\n    }\n\n    /**\n     * Extract the Pulsar IO Sink class from a connector archive.\n     */\n    public static String getIOSinkClass(NarClassLoader narClassLoader) throws IOException {\n        ConnectorDefinition conf = getConnectorDefinition(narClassLoader);\n        if (StringUtils.isEmpty(conf.getSinkClass())) {\n            throw new IOException(\n                    String.format(\"The '%s' connector does not provide a sink implementation\", conf.getName()));\n        }\n\n        try {\n            // Try to load sink class and check it implements Sink interface\n            Class<?> sinkClass = narClassLoader.loadClass(conf.getSinkClass());\n            if (!(Sink.class.isAssignableFrom(sinkClass))) {\n                throw new IOException(\n                        \"Class \" + conf.getSinkClass() + \" does not implement interface \" + Sink.class.getName());\n            }\n        } catch (Throwable t) {\n            Exceptions.rethrowIOException(t);\n        }\n\n        return conf.getSinkClass();\n    }\n\n    public static ConnectorDefinition getConnectorDefinition(File narFile) throws IOException {","sourceCodeStart":77,"sourceCodeEnd":113,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java#L77-L113","documentation":"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.","triggerScenarios":"Calling getIOSinkClass(narClassLoader) on a source-only connector NAR (definition has no sinkClass), e.g. when creating a sink with a source connector archive.","commonSituations":"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.","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"],"exampleFix":"// before\nString cls = ConnectorUtils.getIOSinkClass(sourceOnlyLoader); // throws\n// after\nConnectorDefinition def = ConnectorUtils.getConnectorDefinition(loader);\nif (StringUtils.isNotEmpty(def.getSinkClass())) {\n    String cls = ConnectorUtils.getIOSinkClass(loader);\n}","handlingStrategy":"validation","validationCode":"ConnectorDefinition def = ConnectorUtils.getConnectorDefinition(narClassLoader);\nif (def == null || StringUtils.isEmpty(def.getSinkClass())) {\n    throw new IllegalArgumentException(\"NAR is not a sink connector: \" + def.getName());\n}","typeGuard":"boolean isSinkConnector(ConnectorDefinition def) {\n    return def != null && StringUtils.isNotEmpty(def.getSinkClass());\n}","tryCatchPattern":"try {\n    String sink = ConnectorUtils.getIOSinkClass(loader);\n} catch (IOException e) {\n    if (e.getMessage().contains(\"does not provide a sink implementation\")) {\n        // treat as source-only connector\n    } else { throw e; }\n}","preventionTips":["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"],"tags":["pulsar-io","connector","nar","missing-sink-class"],"backgroundTag":"connector-missing-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"}