KafkaExample.java

package com.softwaremill.kmq;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStoreSupplier;
import org.apache.kafka.streams.processor.TopologyBuilder;
import org.apache.kafka.streams.state.Stores;

import java.util.Properties;

public class StateStoreTest {
    public static void main(String[] args) {
        StateStoreSupplier testStore = Stores.create("store1")
                .withStringKeys()
                .withStringValues()
                .persistent()
                .build();

        TopologyBuilder builder = new TopologyBuilder();

        builder.addSource("source", "topic1")
                .addProcessor("process", TestProcessor::new, "source")
                .addStateStore(testStore, "process");

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "app1");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        KafkaStreams streams = new KafkaStreams(builder, props);
        streams.start();
    }

    public static class TestProcessor implements Processor<String, String> {
        @Override
        public void init(ProcessorContext context) {
            context.getStateStore("store1");
            System.out.println("Initialized");
        }

        @Override
        public void process(String key, String value) {
            System.out.println("Processing " + key + " -> " + value);
        }

        @Override
        public void punctuate(long timestamp) {

        }

        @Override
        public void close() {

        }
    }
}

Hello Adam.. Were you able to resolve this - https://qnalist.com/questions/7980835/kafka-streams-getting-a-state-store-associated-to-a-processor ??? I am hitting the same problem as soon as partition count of the source topic is increased (more than 1)

Thanks!