Kafka流拓扑不同的键，但相同的架构

Question

我有一个Kafka Streams拓扑，其中我加入了5个表，每个表都是在一个主题上创建的，该主题由一些Kafka连接器填充，这些连接器产生KeyValue事件，其中Key是针对相同的Avro模式产生的，但在我的拓扑中，当我加入这些表时，Key似乎不一样，如果它们是Java等于事件。所有这些背后的原因是什么？

它与Confluent Schema Registry集成。

我们已经使用了调试器，并且在调试时看到在不同主题上收到的两个键，但具有相同的值是相等的。但与此同时，如果在主题B之上构建的商店中执行查找，并且该查找是在主题A上接收的键，则它将不匹配某些内容。

fun streamsBuilder(): StreamsBuilder {
    val streamsBuilder = StreamsBuilder()
    val productsStream = streamsBuilder.stream<Key, Aggregate>(streamNameRepository.inputWebshopProductsTopic)
    val productPricesStream = streamsBuilder.stream<Key, PriceVariantsHolder>(streamNameRepository.productsPricesStreamTopic)
    val productsRatingsStream = streamsBuilder.stream<Key, Aggregate>(streamNameRepository.inputProductRatingsTopic)
    val inputProductsStockStream = streamsBuilder.stream<Key, Aggregate>(streamNameRepository.inputProductsStockTopic)

    val productsStockStream =
            inputProductsStockStream.map { key, value -> toKeyValue(key, productStockMapper.aStockQuantity(value)) }
    productsStockStream.to(streamNameRepository.productsStockStreamTopic)

    streamsBuilder.globalTable<Key, StockQuantity>(streamNameRepository.productsStockStreamTopic,
            Materialized.`as`(streamNameRepository.productsStockGlobalStoreTopic))

    val saleProductsTable = productsStream
            .filter { _, aggregate -> aggregate.payload != null }
            .map { key, aggregate -> toKeyValue(key, saleProductMapper.aSaleProduct(aggregate) { productsStockStore().get(Key(it)) }) }
            .mapValues { saleProduct -> log.debug("received $saleProduct"); saleProduct; }
            .groupByKey()
            .reduce({ _, saleProductAvro -> saleProductAvro }, Materialized.`as`(streamNameRepository.saleProductsStoreTopic))

    val productPricesTable = productPricesStream
            .map { key, aggregate -> toKeyValue(key, aggregate) }
            .groupByKey()
            .reduce({ _, price -> price }, Materialized.`as`(streamNameRepository.productsPricesStoreTopic))

    val productsRatingsTable = productsRatingsStream
            .map { key, aggregate -> toKeyValue(key, productRatingMapper.aProductRating(aggregate)) }
            .groupByKey()
            .reduce({ _, aggregate -> aggregate }, Materialized.`as`(streamNameRepository.productsRatingsStoreTopic))

    val productsStockTable = productsStockStream
            .map { key, aggregate -> toKeyValue(key, aggregate) }
            .groupByKey()
            .reduce { _, aggregate -> aggregate }

    val productsInNeedOfVariantStockUpdate = productsInNeedOfVariantStockUpdate(productsStockTable, saleProductsTable)

    saleProductsTable
            .outerJoin(productPricesTable, saleProductMapper::aPricedSaleProduct)
            .outerJoin(productsRatingsTable, saleProductMapper::aRatedSaleProduct)
            .outerJoin(productsStockTable, saleProductMapper::aQuantifiedSaleProduct)
            .outerJoin(productsInNeedOfVariantStockUpdate, saleProductMapper::aSaleProductWithUpdatedVariantStock)
            .toStream()
            .filter { _, saleProductAvro -> saleProductAvro.id != null }
            .mapValues { value -> log.debug("publish {}", value); value; }
            .to(streamNameRepository.outputSaleProductsTopic)

    return streamsBuilder
}

private fun <V> toKeyValue(key: Key, value: V): KeyValue<Key, V> {
    return KeyValue(Key.newBuilder(key).build(), value)
}

Answer 1

如果您与融合模式注册中心集成，每个主题的神奇字节将是不同的，因此连接不会像预期的那样工作(因为键比较发生在字节级...)

这是意料之中的。这个问题偶尔会出现，在Kafka Streams中本地解决（即内置）很棘手，因为Confluent Schema注册表是第三方工具，而Kafka Streams应该与它无关。

不过也有一些解决方法。

一种解决方法是将我们在拓扑中收到的每个键重新映射到一个新的键，现在拓扑中的所有键都使用相同的Avro Schema生成（通过模式ID生成相同的Avro Schema）。

其他的选择(并不真的更好)是“剥离神奇的字节”或者为连接键使用不同的数据类型(例如，一些POJO)。因此，所有这些方法都是相似的。