ReadOnlyKeyValueStore from a KTable using spring-kafka

Question

I am migrating a Kafka Streams implementation which uses pure Kafka apis to use spring-kafka instead as it's incorporated in a spring-boot application.

Everything works fine the Stream, GlobalKTable, branching that I have all works perfectly fine but I am having a hard time incorporating a ReadOnlyKeyValueStore. Based on the spring-kafka documentation here: https://docs.spring.io/spring-kafka/docs/2.6.10/reference/html/#streams-spring

It says:

If you need to perform some KafkaStreams operations directly, you can access that internal KafkaStreams instance by using StreamsBuilderFactoryBean.getKafkaStreams(). You can autowire StreamsBuilderFactoryBean bean by type, but you should be sure to use the full type in the bean definition.

Based on that I tried to incorporate it to my example as in the following fragments below:

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
        Map<String, Object> props = defaultStreamsConfigs();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "quote-stream");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "stock-quotes-stream-group");
        return new KafkaStreamsConfiguration(props);
    }

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME)
    public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder(KafkaStreamsConfiguration defaultKafkaStreamsConfig) {
        return new StreamsBuilderFactoryBean(defaultKafkaStreamsConfig);
    }

...

final GlobalKTable<String, LeveragePrice> leverageBySymbolGKTable = streamsBuilder
            .globalTable(KafkaConfiguration.LEVERAGE_PRICE_TOPIC,
                    Materialized.<String, LeveragePrice, KeyValueStore<Bytes, byte[]>>as("leverage-by-symbol-table")
                            .withKeySerde(Serdes.String())
                            .withValueSerde(leveragePriceSerde));

leveragePriceView = myKStreamsBuilder.getKafkaStreams().store("leverage-by-symbol-table", QueryableStoreTypes.keyValueStore());

But adding the StreamsBuilderFactoryBean(which seems to be needed to get a reference to KafkaStreams) definition causes an error:

The bean 'defaultKafkaStreamsBuilder', defined in class path resource [com/resona/springkafkastream/repository/KafkaConfiguration.class], could not be registered. A bean with that name has already been defined in class path resource [org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.class] and overriding is disabled.

The issue is I don't want to control the lifecycle of the stream that's what I get with the plain Kafka APIs so I would like to get a reference to the default managed one as I want spring to manage it but whenever I try to expose the bean it gives the error. Any ideas on what's the correct approach to that using spring-kafka?

P.S - I am not interested in solutions using spring-cloud-stream I am looking for implementations of spring-kafka.

Answer 1

You don't need to define any new beans; something like this should work...

spring.application.name=quote-stream
spring.kafka.streams.properties.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.properties.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde

@SpringBootApplication
@EnableKafkaStreams
public class So69669791Application {

    public static void main(String[] args) {
        SpringApplication.run(So69669791Application.class, args);
    }

    @Bean
    GlobalKTable<String, String> leverageBySymbolGKTable(StreamsBuilder sb) {
        return sb.globalTable("gkTopic",
                Materialized.<String, String, KeyValueStore<Bytes, byte[]>> as("leverage-by-symbol-table"));
    }

    private ReadOnlyKeyValueStore<String, String> leveragePriceView;

    @Bean
    StreamsBuilderFactoryBean.Listener afterStart(StreamsBuilderFactoryBean sbfb,
            GlobalKTable<String, String> leverageBySymbolGKTable) {

        StreamsBuilderFactoryBean.Listener listener = new StreamsBuilderFactoryBean.Listener() {

            @Override
            public void streamsAdded(String id, KafkaStreams streams) {
                leveragePriceView  = streams.store("leverage-by-symbol-table", QueryableStoreTypes.keyValueStore());
            }

        };
        sbfb.addListener(listener);
        return listener;
    }

    @Bean
    KStream<String, String> stream(StreamsBuilder builder) {
        KStream<String, String> stream = builder.stream("someTopic");
        stream.to("otherTopic");
        return stream;
    }

}