Stack Exchange
KafkaStreams adding more than 1 processor in Topology not working
I have added multiple processors in my topology. Only 1 processor is working. I am able to receive messages from topic1,topic2 and process them using Processor1 and prod output to topic3-sink. Processor2 is not being called in my code.
@Component
public class CustomTopology {
    @PostConstruct
    public void createTopology() {
        Topology topology = new Topology();
        topology.addSource("source", new KeyDeserializer(), new ValueDeserializer(), "topic1", "topic2");
        topology.addProcessor("processor1", new CustomProcessorSupplier1(), "source");
        topology.addProcessor("processor2", new CustomProcessorSupplier2(), "processor1");
        topology.addSink("sink", "topic3-sink", "source", "processor2");
        Properties p = new Properties();
        p.put(APPLICATION_ID_CONFIG, "stream1");
        p.put(BOOTSTRAP_SERVERS_CONFIG, KafkaUtil.getBootStrapServers());
        p.put(DEFAULT_KEY_SERDE_CLASS_CONFIG, KeySerde.class);
        p.put(DEFAULT_VALUE_SERDE_CLASS_CONFIG, ValueSerde.class);
        KafkaStreams streams = new KafkaStreams(topology, p);
        streams.start();
    }
}

class CustomProcessorSupplier1 implements ProcessorSupplier<Integer, Message> {

    @Override
    public Processor<Integer, Message> get() {
        Processor<Integer, Message> processor = new Processor<Integer, Message>() {
            ...

            @Override
            public void process(Integer key, Message value) {
                value.setName(value.getName() + "ProcessB");
            }

            ...
        };
        return processor;
    }
}

class CustomProcessorSupplier2 implements ProcessorSupplier<Integer, Message> {

    @Override
    public Processor<Integer, Message> get() {
        Processor<Integer, Message> processor = new Processor<Integer, Message>() {
           ...

            @Override
            public void process(Integer key, Message value) {
                value.setName(value.getName() + "ansalProcess");
            }

            ...
        };
        return processor;
    }
}


To pass record forward in Processor you have to call ProcessorContext::forward.
This method is overloaded. You can forward all message to all following nodes, but you can also choose subset of nodes to which message will be forwarded.
You have to add to your Processors impementation (CustomProcessorSupplier1 and CustomProcessorSupplier1) context().forward(someKey, someValue);
Your processors doesn't produce message to topic3-sink. Messages appears there because one of parent's nodes is SourceNode (input topics):

topology.addSink("sink", "topic3-sink", "source", "processor2");


Okay, Thanks for the update. 1 doubt , why is message processed using customprocessor1 if I havent used context.forward(..,..)

@R.B., Message is processed by customprocessor1 because your first node - source node -  forward it's to customprocessor1 (topology.addSource("source", new KeyDeserializer(), new ValueDeserializer(), "topic1", "topic2");)

Thanks @wardziniak. 1 last question, If I am adding - topology.addSink("sink", "topic3-sink", "source", "processor2"); I am consuming from source & processor2. so, in sink topic, I am receiving duplicate messages (processed). Is it possible that I can consume msgs directly from source in sink (without processing, bypassing processors) and 1 msg after processing (from processor2)?
Just started with kafka-streaming, so pardon me for silly/basic questions/doubts.

Never mind, find the explanation. There is no copying of data. Once mutated, mutation will persist.

Can you please help me with this 1 too?
https://stackoverflow.com/questions/56336357/running-multiple-topologies-on-same-instance
