@SuppressWarnings("unchecked") @Test public void testKStreamBranch() { KStreamBuilder builder = new KStreamBuilder(); builder.setApplicationId("X"); 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 expectedKey : expectedKeys) { driver.process(topicName, expectedKey, "V" + expectedKey); } assertEquals(3, processors[0].processed.size()); assertEquals(1, processors[1].processed.size()); assertEquals(2, processors[2].processed.size()); }
@SuppressWarnings("unchecked") public KStreamBranch(Predicate<K, V> ... predicates) { this.predicates = predicates; }
KStreamBranch implements ProcessorSupplier<K, V> { @SuppressWarnings("unchecked") public KStreamBranch(Predicate<K, V> ... predicates) { this.predicates = predicates; } }
KStreamBranch implements ProcessorSupplier<K, V> { @SuppressWarnings("unchecked") public KStreamBranch(Predicate<K, V> ... predicates) { this.predicates = predicates; } @SuppressWarnings("unchecked")  KStreamBranch(Predicate<K, V> ... predicates); }
KStreamBranch implements ProcessorSupplier<K, V> { @SuppressWarnings("unchecked") public KStreamBranch(Predicate<K, V> ... predicates) { this.predicates = predicates; } @SuppressWarnings("unchecked")  KStreamBranch(Predicate<K, V> ... predicates); @Override Processor<K, V> get(); }
KStreamBranch implements ProcessorSupplier<K, V> { @SuppressWarnings("unchecked") public KStreamBranch(Predicate<K, V> ... predicates) { this.predicates = predicates; } @SuppressWarnings("unchecked")  KStreamBranch(Predicate<K, V> ... predicates); @Override Processor<K, V> get();  }