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

Question

我有以下

//Config setup
Properties props = ...; //setup

List<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() {

                    }
                };
            }
        });

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

KStreams : How do you get the (originating) topic of a record?

java

apache-kafka

apache-kafka-streams