如何在KafkaStream应用程序中获取partitionId和TopicName [英] How to get partitionId and TopicName in KafkaStream application

问题描述

我们如何从KafkaStream获取主题名称和分区ID.对于任何其他Kafka使用者，我们可以获取主题名称和partitionId，如下所示:

How do we get topic name and partition id from KafkaStream. For any other Kafka consumer we can get topic name and partitionId like following:

    ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {System.out.printf("consumed: key = %s, value = %s, partition id= %s, offset = %s%n",record.key(), record.value(), record.partition(), record.offset());}

不确定如何在KafkaStreams中获取记录引用.

Not sure how to get the record reference in KafkaStreams.

推荐答案

您可以通过Processor API中公开的 ProcessorContext 获取输入记录的元数据.您可以通过 transform()和类似方法将Processor API嵌入DSL.

You can get meta data of input record via the ProcessorContext that is exposed in the Processor API. You can embed the Processor API in the DSL via transform() and similar methods.

查看文档以获取详细信息:

Check out the docs for details: https://docs.confluent.io/current/streams/developer-guide/processor-api.html#accessing-processor-context

这篇关于如何在KafkaStream应用程序中获取partitionId和TopicName的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！