如何忽略Kafka Streams应用程序中从同一主题读写不同事件类型的某些类型的消息

Question

让我们假设Spring Cloud Stream应用程序从Order主题创建KStream。它对ordercreated{“id”：x,“productid”：y,“customerid”：z}事件感兴趣。一旦到达，它将对其进行处理并生成一个输出事件ordershipped{“id”：x,“productID”：y,“customername”：<,“customeraddress”：z}到相同的订单主题。

我面临的问题是，由于Kafka流应用程序从/向同一个主题读写，所以它试图处理自己的写操作，这是没有意义的。

如何防止此应用程序处理它生成的事件？

更新：正如Artem Bilan和sobychako指出的，我曾考虑过使用kstream.filter()，但有些细节让我怀疑如何处理：

interface ShippingKStreamProcessor {
    ...
    @Input("order")
    fun order(): KStream<String, OrderCreated>

    @Output("output")
    fun output(): KStream<String, OrderShipped>

    @StreamListener
    @SendTo("output")
    fun process(..., @Input("order") order: KStream<Int, OrderCreated>): KStream<Int, OrderShipped> {

OrderCreated类：

data class OrderCreated(var id: Int?, var productId: Int?, var customerId: Int?) {
    constructor() : this(null, null, null)
}

OrderShipped类

data class OrderShipped(var id: Int?, var productId: Int?, var customerName: String?, var customerAddress: String?) {
    constructor() : this(null, null, null, null)
}

我使用JSON作为消息格式，因此消息如下所示：

null

Answer 1

我已经接受了布鲁诺的回答，认为这是解决这个问题的有效方法。但是，我想我已经提出了一个更直接/逻辑的方法，使用了用JSONTypeInfo注释的事件结构。

首先，您需要一个用于Order事件的基类，并指定所有子类。请注意，JSON文档中将添加一个type属性，它将帮助Jackson封送/解封DTO：

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
@JsonSubTypes(value = [
    JsonSubTypes.Type(value = OrderCreatedEvent::class, name = "orderCreated"),
    JsonSubTypes.Type(value = OrderShippedEvent::class, name = "orderShipped")
])
abstract class OrderEvent

data class OrderCreatedEvent(var id: Int?, var productId: Int?, var customerId: Int?) : OrderEvent() {
    constructor() : this(null, null, null)
}

data class OrderShippedEvent(var id: Int?, var productId: Int?, var customerName: String?, var customerAddress: String?) : OrderEvent () {
    constructor() : this(null, null, null, null)
}

有了这一点，OrderCreatedEvent对象的生产者将生成如下所示的消息：

orderEvent.filter { _, value -> value is OrderCreatedEvent }
                .map { key, value -> KeyValue(key, value as OrderCreatedEvent) }

完整的KStream逻辑：

@StreamListener
@SendTo("output")
fun process(@Input("input") input: KStream<Int, Customer>, @Input("order") orderEvent: KStream<Int, OrderEvent>): KStream<Int, OrderShippedEvent> {

        val intSerde = Serdes.IntegerSerde()
        val customerSerde = JsonSerde<Customer>(Customer::class.java)
        val orderCreatedSerde = JsonSerde<OrderCreatedEvent>(OrderCreatedEvent::class.java)

        val stateStore: Materialized<Int, Customer, KeyValueStore<Bytes, ByteArray>> =
                Materialized.`as`<Int, Customer, KeyValueStore<Bytes, ByteArray>>("customer-store")
                        .withKeySerde(intSerde)
                        .withValueSerde(customerSerde)

        val customerTable: KTable<Int, Customer> = input.groupByKey(Serialized.with(intSerde, customerSerde))
                .reduce({ _, y -> y }, stateStore)


        return (orderEvent.filter { _, value -> value is OrderCreatedEvent }
                .map { key, value -> KeyValue(key, value as OrderCreatedEvent) }
                .selectKey { _, value -> value.customerId } as KStream<Int, OrderCreatedEvent>)
                .join(customerTable, { orderIt, customer ->
                    OrderShippedEvent(orderIt.id, orderIt.productId, customer.name, customer.address)
                }, Joined.with(intSerde, orderCreatedSerde, customerSerde))
                .selectKey { _, value -> value.id }
                //.to("order", Produced.with(intSerde, orderShippedSerde))
    }

在此过程之后，我将生成一个新消息键：1值：{“type”：“ordershipped”,“id”：1,“productID”：24,“customername”：“anna”,“customeraddress”：“cipress street”}到订单主题中，但这将被流过滤掉。