apache/pulsar · error · IllegalArgumentException

Input Topics cannot be altered

Error message

Input Topics cannot be altered

What it means

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.

Solutions

1. Ensure inputSpecs in the update contains exactly the same topic keys as the existing sink
2. To change input topics, delete the sink and create a new one with the desired topic set
3. Diff the inputSpecs of the update payload against 'pulsar-admin sinks get' output before updating

Example fix

// before
sinkConfig.setInputSpecs(Map.of("persistent://public/default/new-topic", new ConsumerConfig()));
admin.sinks().updateSink(tenant, namespace, sinkConfig, null);
// after
sinkConfig.setInputSpecs(existingInputSpecs); // same topic keys as deployed sink
admin.sinks().updateSink(tenant, namespace, sinkConfig, null);

Set<String> existingTopics = existing.getInputSpecs().keySet();
Set<String> newTopics = newCfg.getInputSpecs() == null ? Set.of() : newCfg.getInputSpecs().keySet();
if (!existingTopics.containsAll(newTopics)) {
    throw new IllegalArgumentException("Input topics are immutable; found new topics: " + newTopics);
}

try {
    admin.sinks().updateSink(tenant, namespace, cfg, null);
} catch (IllegalArgumentException e) {
    if (e.getMessage().contains("Input Topics cannot be altered")) { /* restore original inputSpecs */ }
    else throw e;
}

When it happens

Trigger: Updating a sink with an inputSpecs map containing a topic name absent from the deployed config (added topic), or a renamed topic key.

Common situations: 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.

Related errors

• isRegexPattern for input topic cannot be altered
• Processing Guarantees cannot be altered
• Retain Key Ordering cannot be altered
• Retain Ordering cannot be altered
• Sink Names differ

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

Appendix: source

Thrown at pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java:674

                                .serdeClassName(serdeClassName)
                                .isRegexPattern(false)
                                .build());
            });
        }
        if (newConfig.getTopicToSchemaType() != null) {
            newConfig.getTopicToSchemaType().forEach((topicName, schemaClassname) -> {
                newConfig.getInputSpecs().put(topicName,
                        ConsumerConfig.builder()
                                .schemaType(schemaClassname)
                                .isRegexPattern(false)
                                .build());
            });
        }
        if (!newConfig.getInputSpecs().isEmpty()) {
            SinkConfig finalMergedConfig = mergedConfig;
            newConfig.getInputSpecs().forEach((topicName, consumerConfig) -> {
                if (!existingConfig.getInputSpecs().containsKey(topicName)) {
                    throw new IllegalArgumentException("Input Topics cannot be altered");
                }
                if (consumerConfig.isRegexPattern() != existingConfig.getInputSpecs().get(topicName).isRegexPattern()) {
                    throw new IllegalArgumentException(
                            "isRegexPattern for input topic " + topicName + " cannot be altered");
                }
                finalMergedConfig.getInputSpecs().put(topicName, consumerConfig);
            });
        }
        if (newConfig.getProcessingGuarantees() != null && !newConfig.getProcessingGuarantees()
                .equals(existingConfig.getProcessingGuarantees())) {
            throw new IllegalArgumentException("Processing Guarantees cannot be altered");
        }
        if (newConfig.getConfigs() != null) {
            mergedConfig.setConfigs(newConfig.getConfigs());
        }
        if (newConfig.getSecrets() != null) {
            mergedConfig.setSecrets(newConfig.getSecrets());
        }

View on GitHub (pinned to 820761864e)