KStream-KTable join writing to the KTable: How to sync the join with the ktable write?

Question

我对以下拓扑的行为有一些疑问：

String topic = config.topic();

KTable<UUID, MyData> myTable = topology.builder().table(UUIDSerdes.get(), GsonSerdes.get(MyData.class), topic);

// Receive a stream of various events
topology.eventsStream()
    // Only process events that are implementing MyEvent
    .filter((k, v) -> v instanceof MyEvent)
    // Cast to ease the code
    .mapValues(v -> (MyEvent) v)
    // rekey by data id
    .selectKey((k, v) -> v.data.id)
    .peek((k, v) -> L.info("Event:"+v.action))
    // join the event with the according entry in the KTable and apply the state mutation
    .leftJoin(myTable, eventHandler::handleEvent, UUIDSerdes.get(), EventSerdes.get())
    .peek((k, v) -> L.info("Updated:" + v.id + "-" + v.id2))
    // write the updated state to the KTable.
    .to(UUIDSerdes.get(), GsonSerdes.get(MyData.class), topic);

我的问题发生在我同时收到不同的事件时。由于我的状态突变是由 leftJoin 完成的，然后由 to 方法编写的。如果使用相同的密钥同时接收到事件 1 和 2，我可能会发生以下情况：

event1 joins with state A => state A mutated to state X
event2 joins with state A => state A mutated to state Y
state X written to the KTable topic
state Y written to the KTable topic

因此，状态 Y 没有 event1 的变化，所以我丢失了数据。

这是我所看到的日志（Processing:... 部分是从值连接器内部记录的）：

Event:Event1
Event:Event2
Processing:Event1, State:none
Updated:1-null
Processing:Event2, State:none
java.lang.IllegalStateException: Event2 event received but we don't have data for id 1

Event1 可以被认为是创建事件：它将在 KTable 中创建条目，因此状态是否为空无关紧要。 Event2 虽然需要将其更改应用到现有状态，但它没有找到任何更改，因为第一个状态突变仍未写入 KTable（它仍未被 to方法）

有没有办法确保我的 leftJoin 和对 ktable 的写入是原子完成的？

谢谢

更新和当前解决方案

感谢@Matthias 的回复，我能够使用 Transformer.

找到解决方案

代码如下所示：

那是变形金刚

public class KStreamStateLeftJoin<K, V1, V2> implements Transformer<K, V1, KeyValue<K, V2>> {

    private final String                    stateName;
    private final ValueJoiner<V1, V2, V2>   joiner;
    private final boolean                   updateState;

    private KeyValueStore<K, V2>            state;

    public KStreamStateLeftJoin(String stateName, ValueJoiner<V1, V2, V2> joiner, boolean updateState) {
        this.stateName = stateName;
        this.joiner = joiner;
        this.updateState = updateState;
    }

    @Override
    @SuppressWarnings("unchecked")
    public void init(ProcessorContext context) {
        this.state = (KeyValueStore<K, V2>) context.getStateStore(stateName);
    }

    @Override
    public KeyValue<K, V2> transform(K key, V1 value) {
        V2 stateValue = this.state.get(key); // Get current state
        V2 updatedValue = joiner.apply(value, stateValue); // Apply join
        if (updateState) {
            this.state.put(key, updatedValue); // write new state
        }
        return new KeyValue<>(key, updatedValue);
    }

    @Override
    public KeyValue<K, V2> punctuate(long timestamp) {
        return null;
    }

    @Override
    public void close() {}
}

这是调整后的拓扑：

String topic = config.topic();
String store = topic + "-store";

KTable<UUID, MyData> myTable = topology.builder().table(UUIDSerdes.get(), GsonSerdes.get(MyData.class), topic, store);

// Receive a stream of various events
topology.eventsStream()
    // Only process events that are implementing MyEvent
    .filter((k, v) -> v instanceof MyEvent)
    // Cast to ease the code
    .mapValues(v -> (MyEvent) v)
    // rekey by data id
    .selectKey((k, v) -> v.data.id)
    // join the event with the according entry in the KTable and apply the state mutation
    .transform(() -> new KStreamStateLeftJoin<UUID, MyEvent, MyData>(store, eventHandler::handleEvent, true), store)
    // write the updated state to the KTable.
    .to(UUIDSerdes.get(), GsonSerdes.get(MyData.class), topic);

当我们使用 KTable 的 KV StateStore 并通过 put 方法事件直接在其中应用更改时，事件应该始终选择更新后的状态。一件事我仍然想知道：如果我有连续的高吞吐量事件怎么办。

我们在 KTable 的 KV 存储上执行的放置与在 KTable 的主题中执行的写入之间是否仍然存在竞争条件？

Answer 1

一个KTable被分片成多个物理存储，每个存储仅由一个线程更新。因此，您描述的情况不可能发生。如果您有 2 条具有相同时间戳的记录都更新同一个分片，它们将一个接一个地处理（以偏移顺序）。因此，第二次更新将看到第一次更新后的状态。

所以也许您只是没有正确描述您的场景？

更新

您不能在进行连接时更改状态。因此，期望

event1 joins with state A => state A mutated to state X

错了。独立于任何处理顺序，当event1加入state A时，它将以只读模式访问state A并且不会修改state A。

因此，当 event2 加入时，它会看到与 event1 相同的状态。对于 stream-table 连接，table 状态仅在从 table-input-topic 读取新数据时更新。

如果您想要一个从两个输入更新的共享状态，您需要使用 transform():

构建自定义解决方案

builder.addStore(..., "store-name");
builder.stream("table-topic").transform(..., "store-name"); // will not emit anything downstream
KStream result = builder.stream("stream-topic").transform(..., "store-name");

这将创建一个由两个处理器共享的存储，并且都可以 read/write 随心所欲。因此，对于 table-input，您可以只更新状态而不向下游发送任何内容，而对于 stream-input，您可以进行连接、更新状态并向下游发送结果。

更新 2

关于解决方案，Transformer应用于状态的更新之间不会存在竞争条件，并记录状态更新后的Transformer进程。这部分将在单线程中执行，记录将从输入主题开始按偏移顺序进行处理。因此，可以确保状态更新对以后的记录可用。

KStream-KTable join writing to the KTable: How to sync the join with the ktable write?

KStream-KTable join writing to the KTable: How to sync the join with the ktable write?

apache-kafka-streams