KStreamImpl extends AbstractStream<K> implements KStream<K, V> { @Override public void to(String topic) { to(null, null, null, topic); } KStreamImpl(KStreamBuilder topology, String name, Set<String> sourceNodes); @Override KStream<K, V> filter(Predicate<K, V> predicate); @Override KStream<K, V> filterNot(final Predicate<K, V> predicate); @Override @SuppressWarnings("unchecked") KStream<K1, V> selectKey(final KeyValueMapper<K, V, K1> mapper); @Override KStream<K1, V1> map(KeyValueMapper<K, V, KeyValue<K1, V1>> mapper); @Override KStream<K, V1> mapValues(ValueMapper<V, V1> mapper); @Override void print(); @Override void print(Serde<K> keySerde, Serde<V> valSerde); @Override void writeAsText(String filePath); @Override void writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde); @Override KStream<K1, V1> flatMap(KeyValueMapper<K, V, Iterable<KeyValue<K1, V1>>> mapper); @Override KStream<K, V1> flatMapValues(ValueMapper<V, Iterable<V1>> mapper); @Override @SuppressWarnings("unchecked") KStream<K, V>[] branch(Predicate<K, V>... predicates); static KStream<K, V> merge(KStreamBuilder topology, KStream<K, V>[] streams); @Override KStream<K, V> through(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K, V> partitioner, String topic); @Override void foreach(ForeachAction<K, V> action); @Override KStream<K, V> through(Serde<K> keySerde, Serde<V> valSerde, String topic); @Override KStream<K, V> through(StreamPartitioner<K, V> partitioner, String topic); @Override KStream<K, V> through(String topic); @Override void to(String topic); @Override void to(StreamPartitioner<K, V> partitioner, String topic); @Override void to(Serde<K> keySerde, Serde<V> valSerde, String topic); @SuppressWarnings("unchecked") @Override void to(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K, V> partitioner, String topic); @Override KStream<K1, V1> transform(TransformerSupplier<K, V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames); @Override KStream<K, V1> transformValues(ValueTransformerSupplier<V, V1> valueTransformerSupplier, String... stateStoreNames); @Override void process(final ProcessorSupplier<K, V> processorSupplier, String... stateStoreNames); @Override KStream<K, R> join(
            KStream<K, V1> other,
            ValueJoiner<V, V1, R> joiner,
            JoinWindows windows,
            Serde<K> keySerde,
            Serde<V> thisValueSerde,
            Serde<V1> otherValueSerde); @Override KStream<K, R> join(
            KStream<K, V1> other,
            ValueJoiner<V, V1, R> joiner,
            JoinWindows windows); @Override KStream<K, R> outerJoin(
            KStream<K, V1> other,
            ValueJoiner<V, V1, R> joiner,
            JoinWindows windows,
            Serde<K> keySerde,
            Serde<V> thisValueSerde,
            Serde<V1> otherValueSerde); @Override KStream<K, R> outerJoin(
            KStream<K, V1> other,
            ValueJoiner<V, V1, R> joiner,
            JoinWindows windows); @SuppressWarnings("unchecked") @Override KStream<K, R> leftJoin(
            KStream<K, V1> other,
            ValueJoiner<V, V1, R> joiner,
            JoinWindows windows,
            Serde<K> keySerde,
            Serde<V1> otherValueSerde); @Override KStream<K, R> leftJoin(
            KStream<K, V1> other,
            ValueJoiner<V, V1, R> joiner,
            JoinWindows windows); @SuppressWarnings("unchecked") @Override KStream<K, R> leftJoin(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner); @Override KTable<Windowed<K>, V> reduceByKey(Reducer<V> reducer,
                                                                 Windows<W> windows,
                                                                 Serde<K> keySerde,
                                                                 Serde<V> aggValueSerde); @Override KTable<Windowed<K>, V> reduceByKey(Reducer<V> reducer,
                                                                 Windows<W> windows); @Override KTable<K, V> reduceByKey(Reducer<V> reducer,
                                    Serde<K> keySerde,
                                    Serde<V> aggValueSerde,
                                    String name); @Override KTable<K, V> reduceByKey(Reducer<V> reducer, String name); @Override KTable<Windowed<K>, T> aggregateByKey(Initializer<T> initializer,
                                                                       Aggregator<K, V, T> aggregator,
                                                                       Windows<W> windows,
                                                                       Serde<K> keySerde,
                                                                       Serde<T> aggValueSerde); @Override KTable<Windowed<K>, T> aggregateByKey(Initializer<T> initializer,
                                                                       Aggregator<K, V, T> aggregator,
                                                                       Windows<W> windows); @Override KTable<K, T> aggregateByKey(Initializer<T> initializer,
                                           Aggregator<K, V, T> aggregator,
                                           Serde<K> keySerde,
                                           Serde<T> aggValueSerde,
                                           String name); @Override KTable<K, T> aggregateByKey(Initializer<T> initializer,
                                           Aggregator<K, V, T> aggregator,
                                           String name); @Override KTable<Windowed<K>, Long> countByKey(Windows<W> windows,
                                                                   Serde<K> keySerde); @Override KTable<Windowed<K>, Long> countByKey(Windows<W> windows); @Override KTable<K, Long> countByKey(Serde<K> keySerde, String name); @Override KTable<K, Long> countByKey(String name); static final String JOINTHIS_NAME; static final String JOINOTHER_NAME; static final String LEFTJOIN_NAME; static final String MERGE_NAME; static final String OUTERTHIS_NAME; static final String OUTEROTHER_NAME; static final String SINK_NAME; static final String SOURCE_NAME; }
@Test public void testToWithNullValueSerdeDoesntNPE() { final KStreamBuilder builder = new KStreamBuilder(); final KStream<String, String> inputStream = builder.stream(stringSerde, stringSerde, "input"); inputStream.to(stringSerde, null, "output"); }