{"record":{"id":"63f8fd8aabf426ff","repo":"apache/pulsar","slug":"input-topics-cannot-be-altered-63f8fd","errorCode":null,"errorMessage":"Input Topics cannot be altered","messagePattern":"Input Topics cannot be altered","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java","lineNumber":674,"sourceCode":"                                .serdeClassName(serdeClassName)\n                                .isRegexPattern(false)\n                                .build());\n            });\n        }\n        if (newConfig.getTopicToSchemaType() != null) {\n            newConfig.getTopicToSchemaType().forEach((topicName, schemaClassname) -> {\n                newConfig.getInputSpecs().put(topicName,\n                        ConsumerConfig.builder()\n                                .schemaType(schemaClassname)\n                                .isRegexPattern(false)\n                                .build());\n            });\n        }\n        if (!newConfig.getInputSpecs().isEmpty()) {\n            SinkConfig finalMergedConfig = mergedConfig;\n            newConfig.getInputSpecs().forEach((topicName, consumerConfig) -> {\n                if (!existingConfig.getInputSpecs().containsKey(topicName)) {\n                    throw new IllegalArgumentException(\"Input Topics cannot be altered\");\n                }\n                if (consumerConfig.isRegexPattern() != existingConfig.getInputSpecs().get(topicName).isRegexPattern()) {\n                    throw new IllegalArgumentException(\n                            \"isRegexPattern for input topic \" + topicName + \" cannot be altered\");\n                }\n                finalMergedConfig.getInputSpecs().put(topicName, consumerConfig);\n            });\n        }\n        if (newConfig.getProcessingGuarantees() != null && !newConfig.getProcessingGuarantees()\n                .equals(existingConfig.getProcessingGuarantees())) {\n            throw new IllegalArgumentException(\"Processing Guarantees cannot be altered\");\n        }\n        if (newConfig.getConfigs() != null) {\n            mergedConfig.setConfigs(newConfig.getConfigs());\n        }\n        if (newConfig.getSecrets() != null) {\n            mergedConfig.setSecrets(newConfig.getSecrets());\n        }","sourceCodeStart":656,"sourceCodeEnd":692,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java#L656-L692","documentation":"Thrown by SinkConfigUtils.validateUpdate when the new config's inputSpecs contains a topic that is not present in the existing config's inputSpecs. Input topics of a deployed sink cannot be added or changed via update; only their ConsumerConfig values are merged for already-known topics.","triggerScenarios":"Updating a sink with an inputSpecs map containing a topic name absent from the deployed config (added topic), or a renamed topic key.","commonSituations":"Developer adds a new input topic to the YAML and runs update instead of recreating; topic renamed after a refactor; config generated from scratch omits the original topic list.","solutions":["Ensure inputSpecs in the update contains exactly the same topic keys as the existing sink","To change input topics, delete the sink and create a new one with the desired topic set","Diff the inputSpecs of the update payload against 'pulsar-admin sinks get' output before updating"],"exampleFix":"// before\nsinkConfig.setInputSpecs(Map.of(\"persistent://public/default/new-topic\", new ConsumerConfig()));\nadmin.sinks().updateSink(tenant, namespace, sinkConfig, null);\n// after\nsinkConfig.setInputSpecs(existingInputSpecs); // same topic keys as deployed sink\nadmin.sinks().updateSink(tenant, namespace, sinkConfig, null);","handlingStrategy":"validation","validationCode":"Set<String> existingTopics = existing.getInputSpecs().keySet();\nSet<String> newTopics = newCfg.getInputSpecs() == null ? Set.of() : newCfg.getInputSpecs().keySet();\nif (!existingTopics.containsAll(newTopics)) {\n    throw new IllegalArgumentException(\"Input topics are immutable; found new topics: \" + newTopics);\n}","typeGuard":null,"tryCatchPattern":"try {\n    admin.sinks().updateSink(tenant, namespace, cfg, null);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"Input Topics cannot be altered\")) { /* restore original inputSpecs */ }\n    else throw e;\n}","preventionTips":["Reuse the exact inputSpecs map from the deployed sink when updating","To change topics, delete and recreate the sink","Diff topic sets before issuing an update"],"tags":["pulsar","functions","config-validation","immutable-field"],"backgroundTag":"immutable-config-field-update","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"}