Kafka Streams 转换后有条件地产生消息

Question

我们有一个用例，我们必须将一些消息读入 KStreams，然后转换消息并有条件地将其生成到另一个主题。

在我们的用例中，为了转换对象，我们进行了下游 API 调用。如果 API 调用成功，则生成到 newTopic1，否则生成到 newTopic2。怎样才能达到同样的效果？？

截至目前，我们正在使用 Streams API 提供的 to 方法使用以下样式为新的 Kafka 主题生成丰富的（即转换的对象）。

KStream<String, Customer> transformedStream = sourceKafkaStream
                .mapValues(cust -> {
                    try {
                        logger.info("Hitting to the downstream to fetch additional information, will take few seconds.");
                        Thread.sleep(7000);
                        return RecordEnrichmentAndTransformationBuilder.enrichTheCustomer(cust);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return cust;
                });
                .to('newTopic1', Produced.with(AppSerdes.String(), AppSerdes.serdeForEnrichedCustomer()));

感谢对此的回应。

Answer 1

使用 DSL api，您使用 KStream::filter 或 KStream:to(TopicNameExtractor<K, V> topicExtractor, Produced<K, V> produced)。

如果两种格式相同，示例代码将如下所示：

KStream<String, Customer> transformedStream = sourceKafkaStream
                .mapValues(cust -> {
                    try {
                        logger.info("Hitting to the downstream to fetch additional information, will take few seconds.");
                        Thread.sleep(7000);
                        return RecordEnrichmentAndTransformationBuilder.enrichTheCustomer(cust);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return cust;
                });
                .to((key, value, recordContext) -> topicNameCalculation(key, value), Produced.with(AppSerdes.String(), AppSerdes.serdeForEnrichedCustomer()));

topicNameCalculation(...) 将根据键和值选择正确的主题。

一个通知 一般来说，在 Kafka Streams 中进行外部调用并不是一个好方法。