KStream send record to multiple streams (not Branc

Question

Is there a way to make branch-like operation but to place record in each output stream which predicate evaluates to true? Brach puts record to first match (documentation: A record is placed to one and only one output stream on the first match).

Answer 1

I think you can use something like this:

KStream<String, String> inputStream = builder.stream("input");
List<Predicate<String, String>> predicates = new ArrayList<>(); // <-- list of predicates
List<KStream<String, String>> kStreams = predicates.stream()
        .map(inputStream::branch)
        .map(Arrays::asList)
        .map(listOfOneElementKStreams -> listOfOneElementKStreams.get(0)).collect(Collectors.toList());

Answer 2

You can "broadcast" and filter each stream individually:

KStream stream = ...

stream1 = stream.filter(...);
stream2 = stream.filter(...);
// and so on...

If you use stream variable multiple times, all records are broadcasted to all downstream filters (or any other operator), ie, each filter is executed for each record.