{"record":{"id":"3c1f81ccfe599d4e","repo":"flowable/flowable-engine","slug":"it-is-not-possible-to-auto-create-new-topics-when","errorCode":null,"errorMessage":"It is not possible to auto create new topics when no kafka admin operations have been configured","messagePattern":"It is not possible to auto create new topics when no kafka admin operations have been configured","errorType":"exception","errorClass":"FlowableException","httpStatus":null,"severity":"error","filePath":"modules/flowable-event-registry-spring/src/main/java/org/flowable/eventregistry/spring/kafka/KafkaChannelDefinitionProcessor.java","lineNumber":353,"sourceCode":"                .collect(Collectors.toList());\n    }\n\n    protected static TopicPartitionOffset getTPOForRetryTopics(DestinationTopic.Properties properties, Suffixer suffixer, TopicPartitionOffset tpo) {\n        return new TopicPartitionOffset(suffixer.maybeAddTo(tpo.getTopic()),\n                tpo.getPartition() <= properties.numPartitions() ? tpo.getPartition() : DEFAULT_PARTITION_FOR_MANUAL_ASSIGNMENT);\n    }\n\n    protected static TopicPartitionOffset getTPOForMainTopic(Suffixer suffixer, TopicPartitionOffset tpo) {\n        TopicPartitionOffset newTpo = new TopicPartitionOffset(suffixer.maybeAddTo(tpo.getTopic()),\n                tpo.getPartition(), tpo.getOffset(), tpo.getPosition());\n        newTpo.setRelativeToCurrent(tpo.isRelativeToCurrent());\n        return newTpo;\n    }\n\n    protected Consumer<Collection<String>> getTopicCreationFunction(ResolvedRetryConfiguration retryConfiguration) {\n        if (retryConfiguration.autoCreateTopics) {\n            if (kafkaAdminOperations == null) {\n                throw new FlowableException(\"It is not possible to auto create new topics when no kafka admin operations have been configured\");\n            }\n            return topics -> createNewTopics(topics, retryConfiguration.numPartitions, retryConfiguration.replicationFactor);\n        }\n        return topics -> {};\n    }\n\n    protected void createNewTopics(Collection<String> topics, int numPartitions, short replicationFactor) {\n        kafkaAdminOperations.createOrModifyTopics(topics.stream().map(topic -> new NewTopic(topic, numPartitions, replicationFactor)).toArray(NewTopic[]::new));\n    }\n\n    protected ListenerContainerFactoryConfigurer createListenerContainerFactoryConfigurer(ResolvedRetryConfiguration retryConfiguration, BackOff backOff,\n            DefaultDestinationTopicResolver topicResolver) {\n        DeadLetterPublishingRecovererFactory recovererFactory = new DeadLetterPublishingRecovererFactory(topicResolver);\n\n        KafkaConsumerBackoffManager manager = getOrCreateKafkaConsumerBackoffManager();\n        ListenerContainerFactoryConfigurer factoryConfigurer = new ListenerContainerFactoryConfigurer(manager, recovererFactory, Clock.systemUTC());\n        if (retryConfiguration.hasNoRetryTopic()) {\n            // If we do not have a retry topic, then the retries have to be blocking","sourceCodeStart":335,"sourceCodeEnd":371,"githubUrl":"https://github.com/flowable/flowable-engine/blob/d6d39ce1c69ff244f2d9dc6af756a9b95e865586/modules/flowable-event-registry-spring/src/main/java/org/flowable/eventregistry/spring/kafka/KafkaChannelDefinitionProcessor.java#L335-L371","documentation":"When a Kafka inbound channel has topic auto-creation enabled (retryConfiguration.autoCreateTopics), the processor must create missing Kafka topics before the listener starts. This requires a KafkaAdminOperations bean; if none is configured, FlowableException is thrown from getTopicCreationFunction. The library refuses to silently skip topic creation when it was explicitly requested.","triggerScenarios":"A Kafka inbound/outbound channel model sets autoCreateTopics (e.g. retry topic auto-creation) to true while no KafkaAdminOperations bean is injected into KafkaChannelDefinitionProcessor, and createEndpointConfigurations runs during endpoint registration.","commonSituations":"Spring Boot apps that rely on Spring's KafkaAdmin without exposing it as Flowable's KafkaAdminOperations; missing spring-kafka admin configuration; upgrading Flowable event-registry and enabling topic auto-creation without adding the admin bean; minimal XML/Java config that only sets connection factories.","solutions":["Register a KafkaAdminOperations bean (e.g. a Flowable KafkaAdminOperations bean backed by KafkaAdmin / AdminClient) in the Spring context so the processor can inject it.","Alternatively, disable auto topic creation in the channel/retry configuration so no admin operations are needed.","Pre-create the topics manually (kafka-topics.sh or IaC) and set autoCreateTopics=false.","If using Spring Boot, ensure KafkaAdmin is available and wired; verify the property/bean name matches what KafkaChannelDefinitionProcessor expects."],"exampleFix":"// before\n@Bean\npublic KafkaChannelDefinitionProcessor kafkaChannelDefinitionProcessor(...) {\n    KafkaChannelDefinitionProcessor p = new KafkaChannelDefinitionProcessor();\n    // kafkaAdminOperations never set\n    return p;\n}\n\n// after\n@Bean\npublic KafkaChannelDefinitionProcessor kafkaChannelDefinitionProcessor(KafkaAdmin kafkaAdmin) {\n    KafkaChannelDefinitionProcessor p = new KafkaChannelDefinitionProcessor();\n    p.setKafkaAdminOperations(kafkaAdmin); // admin ops now available for topic creation\n    return p;\n}","handlingStrategy":"validation","validationCode":"if (autoCreateTopics && applicationContext.getBean(KafkaAdminOperations.class) == null) {\n    throw new IllegalStateException(\"autoCreateTopics requires a KafkaAdminOperations bean\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always register a KafkaAdminOperations bean when enabling topic auto-creation","Keep autoCreateTopics=false unless topics are provisioned by the app","Provision topics via IaC and disable runtime auto-creation","Add an application-context smoke test that deploys channel models at boot"],"tags":["kafka","configuration","spring","topic-creation"],"backgroundTag":"missing-required-config","analyzedSha":"d6d39ce1c69ff244f2d9dc6af756a9b95e865586","analyzedAt":"2026-09-11T06:41:19.413Z","contentChangedAt":"2026-09-11T06:41:19.413Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}