@SuppressWarnings("unchecked")
@Test
public void testKStreamBranch() {

    KStreamBuilder builder = new KStreamBuilder();
    Predicate<Integer, String> isEven = new Predicate<Integer, String>() {
        @Override
        public boolean test(Integer key, String value) {
            return (key % 2) == 0;
        }
    };
    Predicate<Integer, String> isMultipleOfThree = new Predicate<Integer, String>() {
        @Override
        public boolean test(Integer key, String value) {
            return (key % 3) == 0;
        }
    };
    Predicate<Integer, String> isOdd = new Predicate<Integer, String>() {
        @Override
        public boolean test(Integer key, String value) {
            return (key % 2) != 0;
        }
    };
    final int[] expectedKeys = new int[]{1, 2, 3, 4, 5, 6};
    KStream<Integer, String> stream;
    KStream<Integer, String>[] branches;
    MockProcessorSupplier<Integer, String>[] processors;
    stream = builder.stream(Serdes.Integer(), Serdes.String(), topicName);
    branches = stream.branch(isEven, isMultipleOfThree, isOdd);
    assertEquals(3, branches.length);
    processors = (MockProcessorSupplier<Integer, String>[]) Array.newInstance(MockProcessorSupplier.class, branches.length);
    for (int i = 0; i < branches.length; i++) {
        processors[i] = new MockProcessorSupplier<>();
        branches[i].process(processors[i]);
    }
    driver = new KStreamTestDriver(builder);
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
    }
    assertEquals(3, processors[0].processed.size());
    assertEquals(1, processors[1].processed.size());
    assertEquals(2, processors[2].processed.size());
}