Invalid topology: Topology has no stream threads and no global threads, must subscribe to at least one source topic or global table

Question

Error I'm getting when I start my spring boot application.

org.springframework.context.ApplicationContextException: Failed to start bean 'defaultKafkaStreamsBuilder'; nested exception is org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.apache.kafka.streams.errors.TopologyException: Invalid topology: Topology has no stream threads and no global threads, must subscribe to at least one source topic or global table. at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:181) ~[spring-context-5.3.28.jar:5.3.28] at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:54) ~[spring-context-5.3.28.jar:5.3.28] at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:356) ~[spring-context-5.3.28.jar:5.3.28] at java.lang.Iterable.forEach(Iterable.java:86) ~[?:2.9 (09-29-2022)] at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:155) ~[spring-context-5.3.28.jar:5.3.28] at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:123) ~[spring-context-5.3.28.jar:5.3.28] at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:937) ~[spring-context-5.3.28.jar:5.3.28] at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:586) ~[spring-context-5.3.28.jar:5.3.28] at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:147) ~[spring-boot-2.7.13.jar:2.7.13] at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:731) ~[spring-boot-2.7.13.jar:2.7.13] at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:408) ~[spring-boot-2.7.13.jar:2.7.13] at org.springframework.boot.SpringApplication.run(SpringApplication.java:307) ~[spring-boot-2.7.13.jar:2.7.13] at org.springframework.boot.SpringApplication.run(SpringApplication.java:1303) ~[spring-boot-2.7.13.jar:2.7.13] at org.springframework.boot.SpringApplication.run(SpringApplication.java:1292) ~[spring-boot-2.7.13.jar:2.7.13] at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:90) ~[?:1.8.0] at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:55) ~[?:1.8.0] at java.lang.reflect.Method.invoke(Method.java:508) ~[?:1.8.0] at org.springframework.boot.devtools.restart.RestartLauncher.run(RestartLauncher.java:50) ~[spring-boot-devtools-2.7.13.jar:2.7.13] Caused by: org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.apache.kafka.streams.errors.TopologyException: Invalid topology: Topology has no stream threads and no global threads, must subscribe to at least one source topic or global table. at org.springframework.kafka.config.StreamsBuilderFactoryBean.start(StreamsBuilderFactoryBean.java:371) ~[spring-kafka-2.9.1.jar:2.9.1] at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178) ~[spring-context-5.3.28.jar:5.3.28] ... 19 more Caused by: org.apache.kafka.streams.errors.TopologyException: Invalid topology: Topology has no stream threads and no global threads, must subscribe to at least one source topic or global table. at org.apache.kafka.streams.processor.internals.TopologyMetadata.getNumStreamThreads(TopologyMetadata.java:215) ~[kafka-streams-3.1.2.jar:?] at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:916) ~[kafka-streams-3.1.2.jar:?] at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:845) ~[kafka-streams-3.1.2.jar:?] at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:751) ~[kafka-streams-3.1.2.jar:?] at org.springframework.kafka.config.StreamsBuilderFactoryBean.start(StreamsBuilderFactoryBean.java:349) ~[spring-kafka-2.9.1.jar:2.9.1] at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178) ~[spring-context-5.3.28.jar:5.3.28] ... 19 more

code:

import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.EnableKafkaStreams;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

@Configuration
@EnableKafka
@EnableKafkaStreams
public class KafkaStreamsConfiguration {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.application.id}")
    private String applicationId;


    @Bean
    public KafkaStreams kafkaStreams(KafkaStreamsConfiguration streamsConfig) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);

        StreamsBuilder builder = new StreamsBuilder();
        KStream < String, String > kStream = builder.stream("retry-events", Consumed.with(Serdes.String(), Serdes.String()));

        // Define a custom ObjectMapper for JSON parsing
        ObjectMapper objectMapper = new ObjectMapper();

        // Use mapValues to parse JSON and extract the timestamp
        kStream = kStream.mapValues(value - > {
            try {
                JsonNode jsonNode = objectMapper.readTree(value);
                long timestamp = jsonNode.get("timestamp").asLong();
                long tenMinutesAgo = System.currentTimeMillis() / 60000; // Current time in minutes
                if (timestamp <= tenMinutesAgo) {
                    return value; // Keep messages that match the criteria
                } else {
                    return null; // Filter out messages that don't match the criteria
                }
            } catch (Exception e) {
                e.printStackTrace();
                return null;
            }
        }).filter((key, value) - > value != null); // Remove messages that failed parsing

        kStream.to("output-topic", Produced.with(Serdes.String(), Serdes.String()));

        KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), props);
        kafkaStreams.start();
        return kafkaStreams;
    }

}

Answer 1

Removing @EnableKafkaStreams makes my application start without any errors.