CODEWITHSUNDEEP
javaapache-kafkaapache-kafka-streams
stackoverflow
stackoverflow

Reputation: 19454

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

I have the following

//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();

Upvotes: 1

Views: 814

Answers (1)

dmkvl
dmkvl

Reputation: 768

You can get a topic name you need using ProcessorContext.topic(). To get an access to ProcessorContext use KStream.process() providing it with an appropriate Processor implementation.

Also you can use 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() {

                    }
                };
            }
        });

Upvotes: 4

Related Questions