You should group your order stream on "customerID" and aggregate all the orders in a list. Result KTable will have <CustomerId, [List of Order]> types events.
Using Interactive queries, you can query the state store,
StreamsBuilder builder = new StreamsBuilder();
KStream<String, Order> orderStream = builder.stream("orders");
KTable<String,ArrayList<Order>> orderTable = orderStream
.groupBy((key,value)-> value .get("customerId"))
.aggregate(()-> new ArrayList<Order>(),
(key,val,agg)-> agg.add(val),
Materialized.as("customer-orders")
.withValueSerde(ArrayListSerde())
);
It will create a materialized view "customer-orders", which you can query via rest endpoint.
You can follow the below link for Exposing KTables as Rest Endpoint :
https://docs.confluent.io/current/streams/developer-guide/interactive-queries.html
The problem of this approach might be, that the
List of Ordersmay grow too large and exceed the maximum message size.Agree, in order to handle that problem, OP needs to change the max.message.bytes or potentially change the keys for optimizing the list. I had the similar issue so had to make more details keys to reduce the list size.
I had to change the code in the answer a bit to make it work. 1. I had to add in Serialized.with for teh groupby so that it would de/serialize the orders correctly 2. ()-> new ArrayList<Order>() didn't work. What worked was .aggregate(ArrayList::new, (newKey,val,agg) -> { agg.add(val); return agg; },