事件采购 - Apache Kafka + Kafka Streams - 如何确保原子性/交易性

Question

我正在使用Apache Kafka Streams评估事件采购，以了解复杂场景的可行性。与关系数据库一样，我遇到过一些案例，原子性/事务性是必不可少的：

购物应用程序有两个服务：

OrderService：有一个带有订单的Kafka Streams商店（OrdersStore）
ProductService：拥有Kafka Streams商店（ProductStockStore）的产品及其库存。

流：

OrderService发布OrderCreated事件（带有productId，orderId，userId信息）
ProductService获取OrderCreated事件并查询其KafkaStreams Store（ProductStockStore）以检查产品是否有库存。如果有库存，它会发布OrderUpdated事件（也包含productId，orderId，userId info）

关键是这个事件将由ProductService Kafka Stream监听，它会处理它以减少库存，到目前为止一直很好。

但是，想象一下：

客户1下订单，订单1（产品库存为1）
客户2同时放置另一个订单，order2，用于同一产品（库存仍为1）
ProductService处理order1并发送消息OrderUpdated以减少库存。此消息放在order2 - > OrderCreated之后的主题中
ProductService处理order2-OrderCreated并发送消息OrderUpdated以再次减少库存。这是不正确的，因为它会引入不一致（库存现在应为0）。

显而易见的问题是，当我们处理第一个OrderUpdated事件时，我们的物化视图（商店）应该直接更新。然而，更新Kafka Stream Store的唯一方法（我知道）是发布另一个由Kafka Stream处理的事件（OrderUpdated）。这样我们就无法以事务方式执行此更新。

我希望能够处理这样的场景。

更新：我将尝试澄清问题的有问题：

ProductService有一个Kafka Streams商店，产品与库存(productId=1, quantity=1)

OrderService在订单主题上发布两个OrderPlaced事件：

Event1 (key=product1, productId=product1, quantity=1, eventType="OrderPlaced")
Event2 (key=product1, productId=product1, quantity=1, eventType="OrderPlaced")

ProductService在订单主题上有一个消费者。为简单起见，我们假设一个分区来确保消息按顺序消耗。此使用者执行以下逻辑：

if("OrderPlaced".equals(event.get("eventType"))){

    Order order = new Order();
    order.setId((String)event.get("orderId"));
    order.setProductId((Integer)(event.get("productId")));
    order.setUid(event.get("uid").toString());

    // QUERY PRODUCTSTOCK TO CHECK AVAILABILITY
    Integer productStock = getProductStock(order.getProductId());

    if(productStock > 0) {
        Map<String, Object> event = new HashMap<>();
        event.put("name", "ProductReserved");
        event.put("orderId", order.getId());
        event.put("productId", order.getProductId());

        // WRITES A PRODUCT RESERVED EVENT TO orders topic
        orderProcessor.output().send(MessageBuilder.withPayload(event).build(), 500);
    }else{
        //XXX CANCEL ORDER
    }
}

ProductService还有一个Kafka Streams处理器，负责更新库存：

KStream<Integer, JsonNode> stream = kStreamBuilder.stream(integerSerde, jsonSerde, "orders");
stream.xxx().yyy(() -> {...}, "ProductsStock");

Event1将首先被处理，因为仍有1个可用产品，它将生成ProductReserved事件。

现在，它已经是Event2了。如果ProductService使用者消耗它，那么在ProductService Kafka Streams Processor处理Event1生成的ProductReseved事件之前，消费者仍会看到product1的ProductStore库存为1，为Event2生成ProductReserved事件，然后在系统中产生不一致。

Answer 1

另一答案