Stack Exchange
Not able to see any messages in the Kafka Ktable with Spring Cloud streamng
I have written a Kafka streams application using Spring Cloud streaming API and not able to see any messages in the KTable. I am not able to trace the issue. Any pointers or help appreciated.
Below is the code
import java.util.function.Function;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.state.KeyValueStore;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class KafkaStreamsSample {

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

        //bin/confluent local produce user -- --property parse.key=true --property key.separator=~
        //2~{"id": "2", "name": "john", "age": 43}
        //1~{"id": "1", "name": "bob", "age": 44}
        //3~{"id": "3", "name": "peter", "age": 45}
        //4~{"id": "4", "name": "mark", "age": 46}
        //2~{"id": "2", "name": "john", "age": 99}
        //3~{"id": "3", "name": "paul", "age": 98}

    public static class KStreamToTableJoinApplication {

        @Bean
        public Function<KStream<String, User>, KTable<String, User>> process() {


                return input -> input
                .groupByKey()
                .reduce((aggValue, newValue) -> newValue, Materialized.as ("allusers"));
        }
    }
}

application.yml
spring.application.name: stream-global-sample
spring.cloud.stream.bindings.process-in-0:
  destination: user
spring.cloud.stream.bindings.process-out-0:
  destination: usertable
spring.cloud.stream.kafka.streams.bindings.process-out-0:
  producer:
    materializedAs: allusers


Using the Spring Cloud Stream Kafka Streams binder, you cannot have the outbound as a KTable. It has to be a KStream. Change your signature to this: public Function<KStream<String, User>, KStream<String, User>>. Then, call toStream() on your reduce call. This will give you a KStream to return. That should allow you to see the output in the outbound topic. The result from your reduce operation is materialized into a state store though through a KTable. So, if you wish, you can directly query that state store through interactive queries.  

Thank you @sobychacko. Yes, as stream I am able to see the output. I suspected the configuration might be wrong for the state store. Is this the right approach for creating a cache of db changes?

Not sure what you mean by the DB changes here without knowing the actual use case, but the changes are persisted as key/value pairs into the state store when you call the reduce method and materialize into the named state store.

Apologies for the limited info provided. Basically I was trying to simulate an usecase where the input event is a database change which is grouped by key and stored it in state store as cache of the changed data. But the above code doesn't work when the input event is partial. Does ktable support partial updates? Any suggestions appreciated. Thanks

what do you mean by partial? I suggest starting a new thread on this as the original issue is already resolved.

Thanks, I have started another question here https://stackoverflow.com/q/58960806/12382975

When calling toStream() , how do you customize the serdes as I am not using the default

Sorry, I missed it, are you still looking for an answer for that?
