flowable/flowable-engine · error · FlowableException

It is not possible to auto create new topics when no kafka…

Error message

It is not possible to auto create new topics when no kafka admin operations have been configured

What it means

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.

Solutions

1. Register a KafkaAdminOperations bean (e.g. a Flowable KafkaAdminOperations bean backed by KafkaAdmin / AdminClient) in the Spring context so the processor can inject it.
2. Alternatively, disable auto topic creation in the channel/retry configuration so no admin operations are needed.
3. Pre-create the topics manually (kafka-topics.sh or IaC) and set autoCreateTopics=false.
4. If using Spring Boot, ensure KafkaAdmin is available and wired; verify the property/bean name matches what KafkaChannelDefinitionProcessor expects.

Example fix

// before
@Bean
public KafkaChannelDefinitionProcessor kafkaChannelDefinitionProcessor(...) {
    KafkaChannelDefinitionProcessor p = new KafkaChannelDefinitionProcessor();
    // kafkaAdminOperations never set
    return p;
}

// after
@Bean
public KafkaChannelDefinitionProcessor kafkaChannelDefinitionProcessor(KafkaAdmin kafkaAdmin) {
    KafkaChannelDefinitionProcessor p = new KafkaChannelDefinitionProcessor();
    p.setKafkaAdminOperations(kafkaAdmin); // admin ops now available for topic creation
    return p;
}

if (autoCreateTopics && applicationContext.getBean(KafkaAdminOperations.class) == null) {
    throw new IllegalStateException("autoCreateTopics requires a KafkaAdminOperations bean");
}

When it happens

Trigger: 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.

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

Understand the failure class

Background: "X is required", "must be set", "cannot be empty": the missing-required-config error family, from Vertex AI project/location to WeChat keys — this error's family across 18 libraries.

Related errors

• At least one of topics, topicPartitions or topicPattern…
• Could not resolve the KafkaListenerContainerFactory to use…
• Only one of topics, topicPartitions or topicPattern must…
• Topics or topicPartitions must be provided but not both for
• ackMode in definition

AI-assisted analysis of flowable/flowable-engine@d6d39ce1c6 (2026-09-11). Data as JSON: /api/errors/3c1f81ccfe599d4e. Report an issue: GitHub.

Appendix: source

Thrown at modules/flowable-event-registry-spring/src/main/java/org/flowable/eventregistry/spring/kafka/KafkaChannelDefinitionProcessor.java:353

                .collect(Collectors.toList());
    }

    protected static TopicPartitionOffset getTPOForRetryTopics(DestinationTopic.Properties properties, Suffixer suffixer, TopicPartitionOffset tpo) {
        return new TopicPartitionOffset(suffixer.maybeAddTo(tpo.getTopic()),
                tpo.getPartition() <= properties.numPartitions() ? tpo.getPartition() : DEFAULT_PARTITION_FOR_MANUAL_ASSIGNMENT);
    }

    protected static TopicPartitionOffset getTPOForMainTopic(Suffixer suffixer, TopicPartitionOffset tpo) {
        TopicPartitionOffset newTpo = new TopicPartitionOffset(suffixer.maybeAddTo(tpo.getTopic()),
                tpo.getPartition(), tpo.getOffset(), tpo.getPosition());
        newTpo.setRelativeToCurrent(tpo.isRelativeToCurrent());
        return newTpo;
    }

    protected Consumer<Collection<String>> getTopicCreationFunction(ResolvedRetryConfiguration retryConfiguration) {
        if (retryConfiguration.autoCreateTopics) {
            if (kafkaAdminOperations == null) {
                throw new FlowableException("It is not possible to auto create new topics when no kafka admin operations have been configured");
            }
            return topics -> createNewTopics(topics, retryConfiguration.numPartitions, retryConfiguration.replicationFactor);
        }
        return topics -> {};
    }

    protected void createNewTopics(Collection<String> topics, int numPartitions, short replicationFactor) {
        kafkaAdminOperations.createOrModifyTopics(topics.stream().map(topic -> new NewTopic(topic, numPartitions, replicationFactor)).toArray(NewTopic[]::new));
    }

    protected ListenerContainerFactoryConfigurer createListenerContainerFactoryConfigurer(ResolvedRetryConfiguration retryConfiguration, BackOff backOff,
            DefaultDestinationTopicResolver topicResolver) {
        DeadLetterPublishingRecovererFactory recovererFactory = new DeadLetterPublishingRecovererFactory(topicResolver);

        KafkaConsumerBackoffManager manager = getOrCreateKafkaConsumerBackoffManager();
        ListenerContainerFactoryConfigurer factoryConfigurer = new ListenerContainerFactoryConfigurer(manager, recovererFactory, Clock.systemUTC());
        if (retryConfiguration.hasNoRetryTopic()) {
            // If we do not have a retry topic, then the retries have to be blocking

View on GitHub (pinned to d6d39ce1c6)