CDIStreamProcessor.java

package net.wessendorf;

import net.wessendorf.kafka.cdi.annotation.KafkaStream;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;

public class StreamProcessor {

    @KafkaStream(input = "input_topic", output = "output_topic")
    public KStream transformer(final KStream<String, String> source) {

        return ource.filter((key, value) -> value.equals("Success"))
                .groupByKey()
                .count("successMessagesStore")
                toStream();
    }
}