KStreams：如何获取记录的（原始）主题？

Question

我有以下//Config setupProperties props = ...; //setupList<String> topicList = Arrays.asList({"A", "B", "C"});StreamBuilder builder = new StreamBuilder();KStream<String, String> source = builder.stream(topicList);source  .map((k,v) -> {    //How can i get the topic of the record here  })  .to((k,v,r) -> {//busy code for topic routing});new KafkaStream(builder.build(), properties).start();

Answer 1

您可以使用ProcessorContext.topic()获取所需的主题名称。要访问 ProcessorContext，请使用 KStream.process() 为其提供适当的Processor实现。

您也可以使用 KStream.transform()：

KStream<InputKeyType, InputValueType> stream2 = stream.transform(new TransformerSupplier<InputKeyType, InputValueType, KeyValue<OutputKeyType, OutputValueType>>() {

            @Override

            public Transformer<InputKeyType, InputValueType, KeyValue<OutputKeyType, OutputValueType>> get() {

                return new Transformer<InputKeyType, InputValueType, KeyValue<OutputKeyType, OutputValueType>>() {

                    private ProcessorContext context;

                    @Override

                    public void init(ProcessorContext context) {

                        this.context = context;

                    }

                    @Override

                    public KeyValue<OutputKeyType, OutputValueType> transform(InputKeyType key, InputValueType value) {

                        this.context.topic() // topic name you need

                        // logic here

                        return new KeyValue<>(OutputKeyType key, OutputValueType value);

                    }

                    @Override

                    public void close() {

                    }

                };

            }

        });