Kafka Streams DSL中如何查询state store实现消费者幂等性

Question

我正在处理重复消息可能到达消费者（KStream 应用程序）的场景。要使用典型案例，我们假设它是一个 OrderCreatedEvent，并且 KStream 具有处理订单的逻辑。该活动有一个订单 ID，可以帮助我识别重复的消息。

我想做的是：

1) 将每个订单添加到持久状态存储

2) 在处理KStream中的消息时，查询state store检查消息是否已经收到，在这种情况下什么都不做。

        val persistentKeyValueStore = Stores.persistentKeyValueStore("order-store")

        val stateStore: Materialized<Int, Order, KeyValueStore<Bytes, ByteArray>> =
                Materialized.`as`<Int, Order>(persistentKeyValueStore)
                        .withKeySerde(intSerde)
                        .withValueSerde(orderSerde)

        val orderTable: KTable<Int, Order> = input.groupByKey(Serialized.with(intSerde, orderSerde))
                .reduce({ _, y -> y }, stateStore)

        var orderStream: KStream<Int, Order> = ...

        orderStream.filter { XXX }
                   .map { key, value -> 
                      processingLogic()
                      KeyValue(key, value)
                   }...

在 filter { XXX } 位中，我想查询状态存储检查订单 ID 是否存在（假设订单用作键值存储的键），过滤掉已处理的订单（存在在国营商店）。

我的第一个问题是：如何在 KStream DSL 中查询状态存储，例如在过滤操作里面。

第二个问题：在这种情况下，我如何处理新的（之前未处理过的消息）的到来？如果 KTable 在 orderStream KStream 执行之前将订单持久化到状态存储，则消息将已经在存储中。只有在处理完成后才应添加它们。我怎样才能做到这一点？很可能我不应该为它使用 KTable，而是像这样的东西：

           orderStream.filter { keystore.get(key) == null }
                   .map { key, value -> 
                       processingLogic()
                       KeyValue(key, value)
                   }
                   .foreach { key, value -> 
                       keystore.put(key, value); 
                   }

Answer 1

按照 Matthias 的指示，我是这样实现的：

DeduplicationTransformer

package com.codependent.outboxpattern.operations.stream

import com.codependent.outboxpattern.account.TransferEmitted
import org.apache.kafka.streams.KeyValue
import org.apache.kafka.streams.kstream.Transformer
import org.apache.kafka.streams.processor.ProcessorContext
import org.apache.kafka.streams.state.KeyValueStore
import org.slf4j.LoggerFactory


@Suppress("UNCHECKED_CAST")
class DeduplicationTransformer : Transformer<String, TransferEmitted, KeyValue<String, TransferEmitted>> {

    private val logger = LoggerFactory.getLogger(javaClass)
    private lateinit var dedupStore: KeyValueStore<String, String>
    private lateinit var context: ProcessorContext

    override fun init(context: ProcessorContext) {
        this.context = context
        dedupStore = context.getStateStore(DEDUP_STORE) as KeyValueStore<String, String>
    }

    override fun transform(key: String, value: TransferEmitted): KeyValue<String, TransferEmitted>? {
        return if (isDuplicate(key)) {
            logger.warn("****** Detected duplicated transfer {}", key)
            null
        } else {
            logger.warn("****** Registering transfer {}", key)
            dedupStore.put(key, key)
            KeyValue(key, value)
        }
    }

    private fun isDuplicate(key: String) = dedupStore[key] != null

    override fun close() {
    }
}

FraudKafkaStreams配置

const val DEDUP_STORE = "dedup-store"

@Suppress("UNCHECKED_CAST")
@EnableBinding(TransferKafkaStreamsProcessor::class)
class FraudKafkaStreamsConfiguration(private val fraudDetectionService: FraudDetectionService) {

    private val logger = LoggerFactory.getLogger(javaClass)

    @KafkaStreamsStateStore(name = DEDUP_STORE, type = KafkaStreamsStateStoreProperties.StoreType.KEYVALUE)
    @StreamListener
    @SendTo(value = ["outputKo", "outputOk"])
    fun process(@Input("input") input: KStream<String, TransferEmitted>): Array<KStream<String, *>>? {
        val fork: Array<KStream<String, *>> = input
                .transform(TransformerSupplier { DeduplicationTransformer() }, DEDUP_STORE)
                .branch(Predicate { _: String, value -> fraudDetectionService.isFraudulent(value) },
                        Predicate { _: String, value -> !fraudDetectionService.isFraudulent(value) }) as Array<KStream<String, *>>
                 ...

Kafka Streams DSL中如何查询state store实现消费者幂等性

How to query the state store in the Kafka Streams DSL to implement consumer idempotency

apache-kafka

apache-kafka-streams