[KAFKA-5488] KStream.branch should not return a Array of streams we have to access by known index

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 2.8.0
• Component/s: streams
• Labels: kip
• Assignee: Ivan Ponomarev
• Reporter: Marcel "childNo͡.de" Trautwein
• Created: 21/Jun/17 13:25
• Updated: 05/Feb/21 10:29
• Resolved: 05/Feb/21 00:24

Description

KIP-418: https://cwiki.apache.org/confluence/display/KAFKA/KIP-418%3A+A+method-chaining+way+to+branch+KStream

long story short: it's a mess to get a KStream<>[] out from KStream<>branch(Predicate<>...). It breaks the fluent API and it produces bad code which is not that good to maintain since you have to know the right index for an unnamed branching stream.

Example

import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.KStream;

public class StreamAppWithBranches {
    public static void main(String... args) {
        KStream<byte[], EventType>[] branchedStreams= new KStreamBuilder()
        .<byte[], EventType>stream("eventTopic")
        .branch(
            (k, v) -> EventType::validData
            (k, v) -> true
        );
        
        branchedStreams[0]
        .to("topicValidData");
        
        branchedStreams[1]
        .to("topicInvalidData");
    }
}

Quick idea, s.th. like void branch(final BranchDefinition<Predicate<>, Consumer<KStream<>>>... branchPredicatesAndHandlers); where you can write branches/streams code nested where it belongs to

so it would be possible to write code like

        new KStreamBuilder()
        .<byte[], EventType>stream("eventTopic")
        .branch(
            Branch.create(
                (k, v) -> EventType::validData,
                stream -> stream.to("topicValidData")
            ),
            Branch.create(
                (k, v) -> true,
                stream -> stream.to("topicInvalidData")
            )
        );

I'll go forward to evaluate some ideas:
https://gitlab.com/childno.de/apache_kafka/snippets/1665655

Issue Links

• fixes: KAFKA-8296 Kafka Streams branch method raises type warnings
• relates to: KAFKA-8296 Kafka Streams branch method raises type warnings
• supercedes: KAFKA-8651 Add a #branch overload that takes a Map of predicate names to predicates
• links to: GitHub Pull Request #6164
• links to: GitHub Pull Request #9107
• mentioned in: Page Loading...
• mentioned in: Page Loading...