Splits a KStream into one or more branches, based on the provided predicate. Each branch is a KStream object. The stream processor node evaluates the predicates in order and places the records only on the first branch with a predicate that matches. If there is no match, then the stream processor node drops the record. You use this method if you want to apply different processing for the records based on certain criteria. For example, in a stream of payments, you might need to process payments greater than 1000 USD in a different way than payments below that number.

orders.split() 
    .branch( 
        (key, order) -> order.amount < 100, 
        Branched.withConsumer( 
            this::doSomething
        )
    )
    .branch(
        (key, order) -> order.amount > 9000,
        Branched.withConsumer(
            this::doAnotherThing
        )
    );

The preceding example splits the orders stream into two branches based on the value of the amount field. The first branch contains all the orders with an amount value lower than 100, and uses the doSomething method to process the branch. The second branch contains all the orders with an amount value greater than 9000, and uses the doAnotherThing method to process the branch. The stream processor drops any order with an amount value out of the defined ranges.

Evaluates a predicate for each element of the stream, and creates a new KStream with all the elements that satisfy the predicate. You use this transformation when you want to discard some stream elements based on certain condition. For example, if you want to keep only a number greater than 99999 in a stream of numbers.

orders.filter( 
    (key, order) -> order.amount > 99999 
);

The preceding example creates a new stream that contains all the orders with an amount value greater than 99999.

Evaluates a predicate for each element of the stream, and creates a new KStream with all the elements that do not satisfy the predicate.

orders.filterNot( 
    (key, order) -> order.amount > 99999 
);

The preceding example creates a new stream that contains all the orders with an amount value lower or equal to 99999.

A terminal operation that allows you to perform an action on each record of the stream.

orders.foreach( 
    (key, order) -> {
        log(order); 
    }
);

The preceding example loops through the records in the orders stream, and executes the log method on each one of the events.

Creates a new KStream in which the elements are transformations of each one of the records of another stream. This operation changes the keys, and marks the stream for data repartitioning. That means that the next grouping or join operations will repartition the records. The mapValues operation does not cause data repartitioning.

orders.map( 
    (key, order) -> { 
        return new KeyValue<>(
            order.id,
            new OrderWasCreatedInEur(order.id, order.amount * 1.15)
        );
    }
);

The preceding example transforms the records available in the orders stream, and transforms each one of them into a new OrderWasCreatedInEur event.