Hi Chris,
*"if I understand your problem correctly, the issue is that you need
todecrement the stock count when you reserve it, rather than splitting it*
*into a second phase."*
That's exactly the problem, I would need to:
1) Read the OrderPlaced event from Kafka in ProductService...
2) ...query the ProductsStock store to check availability...
3) ...update the Product in the same phase (OrderPlacedEvent processing)
4) ...publish a ProductReserved message
// 1) Read the OrderPlaced event...
@StreamListener(OrderProcessor.INPUT)
public void handleOrder(Map<String, Object> event){
logger.info("Event {}", event);
if("OrderPlaced".equals(event.get("name"))){
Order order = new Order();
order.setId((String)event.get("orderId"));
order.setProductId((Integer)(event.get("productId")));
order.setUid(event.get("uid").toString());
...
// 2) Query the ProductsStockStore...
Integer productStock =
getProductStock(order.getProductId());
if(productStock != null && productStock > 0) {
// 3) Update the ProductsStockStore
???
// 4) Publish a new message. No problem here
}
@Override
public Integer getProductStock(Integer id) {
KafkaStreams streams = kStreamBuilderFactoryBean.getKafkaStreams();
ReadOnlyKeyValueStore<Integer, Integer> keyValueStore =
streams.store("ProductsStock", QueryableStoreTypes.keyValueStore());
return keyValueStore.get(id);
}
However the only way I know of updating the store is publishing a new event
ProductReserved that will be processed by the KStream as a separated step
(new Kafka message):
Map<String, Object> event = new HashMap<>();
event.put("name", "ProductReserved");
event.put("orderId", order.getId());
event.put("productId", order.getProductId());
event.put("quantity", -1);
// 3) Update the ProductStore
orderProcessor.output().send(MessageBuilder.withPayload(event).build(),
500);
This is the separated KStream config notice // 3) where the update takes
place:
@Configuration
public class KStreamsConfig {
@Bean
public KStream<?, ?> kStream(KStreamBuilder kStreamBuilder,
KStreamBuilderFactoryBean kStreamBuilderFactoryBean) {
Serde<Integer> integerSerde = Serdes.Integer();
final Serializer<JsonNode> jsonSerializer = new
JsonSerializer();
final Deserializer<JsonNode> jsonDeserializer = new
JsonDeserializer();
final Serde<JsonNode> jsonSerde =
Serdes.serdeFrom(jsonSerializer, jsonDeserializer);
KStream<Integer, JsonNode> stream = kStreamBuilder.stream(integerSerde,
jsonSerde, "orders");
// 3) Update the ProductStore
stream.filter( (key, value) -> value != null &&
value.get("name").asText().equals("ProductReserved"))
.map( (key, value) -> {
return new KeyValue<>(value.get("productId").asInt(),
value.get("quantity").asInt());
}).groupByKey().reduce( (v1, v2) -> v1 + v2, "ProductsStock");
return stream;
}
}
I've had a look at the StateStoresInTheDSLIntegrationTest.java
<https://github.com/confluentinc/examples/blob/master/kafka-streams/src/test/java/io/confluent/examples/streams/StateStoresInTheDSLIntegrationTest.java>
but
I still don't get how to integrate the update step in // 2). No idea how I
can do all this in the same phase:
- Consume a message
- Query a KStreams store
- Update the KStreams store
- Publish a ProductReserved message.
Could you please outline the necessary code to do it?
Thank you so much.
Jose
