{"record":{"id":"0b150a1cc8748e5c","repo":"apache/pulsar","slug":"invalid-sink-type-s-available-sinks-are-s","errorCode":null,"errorMessage":"Invalid sink type '%s' -- Available sinks are: %s","messagePattern":"Invalid sink type '(.+?)' -- Available sinks are: (.+?)","errorType":"validation","errorClass":"ParameterException","httpStatus":null,"severity":"error","filePath":"pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java","lineNumber":657,"sourceCode":"                    && !sinkConfig.getArchive().startsWith(Utils.BUILTIN)) {\n                if (!new File(sinkConfig.getArchive()).exists()) {\n                    throw new IllegalArgumentException(String.format(\"Sink Archive file %s does not exist\",\n                            sinkConfig.getArchive()));\n                }\n            }\n        }\n\n        protected String validateSinkType(String sinkType) throws IOException {\n            Set<String> availableSinks;\n            try {\n                availableSinks = getAdmin().sinks().getBuiltInSinks().stream()\n                        .map(ConnectorDefinition::getName).collect(Collectors.toSet());\n            } catch (PulsarAdminException e) {\n                throw new IOException(e);\n            }\n\n            if (!availableSinks.contains(sinkType)) {\n                throw new ParameterException(\n                        \"Invalid sink type '\" + sinkType + \"' -- Available sinks are: \" + availableSinks);\n            }\n\n            // Source type is a valid built-in connector type\n            return \"builtin://\" + sinkType;\n        }\n    }\n\n    /**\n     * Sink level command.\n     */\n    @Getter\n    abstract class SinkCommand extends BaseCommand {\n        @Option(names = \"--tenant\", description = \"The sink's tenant\")\n        protected String tenant;\n\n        @Option(names = \"--namespace\", description = \"The sink's namespace\")\n        protected String namespace;","sourceCodeStart":639,"sourceCodeEnd":675,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java#L639-L675","documentation":"CmdSinks.validateSinkType checked the requested sink type against the built-in connectors returned by the broker/functions worker and found no match; the CLI lists the available sink types in the error.","triggerScenarios":"Running `pulsar-admin sinks create --sink-type xyz` where 'xyz' is not in the connector registry returned by the admin API (GET /connectors) on the broker.","commonSituations":"Connector NAR not deployed to the broker's connectors directory; connector name casing mismatch; using a source connector name for a sink; broker deployed without bundled connectors.","solutions":["Pick a sink type from the listed available sinks","Provide a --archive (NAR) instead of relying on a builtin type"],"exampleFix":"// before\n--sink-type cassandra-sink\n// after\n--sink-type cassandra","handlingStrategy":"validation","validationCode":"const available = execSync('pulsar-admin sinks available-sinks').toString().split('\\n');\nif (!available.includes(sinkType)) throw new Error(`Unknown sink type: ${sinkType}. Available: ${available}`);","typeGuard":"const isValidSinkType = (t, available) => available.includes(t);","tryCatchPattern":"try {\n  // run sinks create --sink-type X\n} catch (e) {\n  if (e.message.includes('Invalid sink type')) { /* list available-sinks and correct */ }\n  throw e;\n}","preventionTips":["Run `pulsar-admin sinks available-sinks` to discover valid names","Ensure connector NARs are installed on the brokers","Match the connector name exactly, including casing"],"tags":["cli","pulsar-io","connector"],"backgroundTag":"invalid-connector-type","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}