@Test(expected = NullPointerException.class) public void shouldNotAllowNullTransformSupplierOnTransform() throws Exception { testStream.transform(null); }
@Override public <K1, V1> KStream<K1, V1> transform(TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames) { Objects.requireNonNull(transformerSupplier, "transformerSupplier can't be null"); String name = topology.newName(TRANSFORM_NAME); topology.addProcessor(name, new KStreamTransform<>(transformerSupplier), this.name); topology.connectProcessorAndStateStores(name, stateStoreNames); return new KStreamImpl<>(topology, name, sourceNodes, true); }
KStreamImpl extends AbstractStream<K> implements KStream<K, V> { @Override public <K1, V1> KStream<K1, V1> transform(TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames) { Objects.requireNonNull(transformerSupplier, "transformerSupplier can't be null"); String name = topology.newName(TRANSFORM_NAME); topology.addProcessor(name, new KStreamTransform<>(transformerSupplier), this.name); topology.connectProcessorAndStateStores(name, stateStoreNames); return new KStreamImpl<>(topology, name, sourceNodes, true); } }
KStreamImpl extends AbstractStream<K> implements KStream<K, V> { @Override public <K1, V1> KStream<K1, V1> transform(TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames) { Objects.requireNonNull(transformerSupplier, "transformerSupplier can't be null"); String name = topology.newName(TRANSFORM_NAME); topology.addProcessor(name, new KStreamTransform<>(transformerSupplier), this.name); topology.connectProcessorAndStateStores(name, stateStoreNames); return new KStreamImpl<>(topology, name, sourceNodes, true); } KStreamImpl(KStreamBuilder topology, String name, Set<String> sourceNodes,
                       boolean repartitionRequired); }
KStreamImpl extends AbstractStream<K> implements KStream<K, V> { @Override public <K1, V1> KStream<K1, V1> transform(TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames) { Objects.requireNonNull(transformerSupplier, "transformerSupplier can't be null"); String name = topology.newName(TRANSFORM_NAME); topology.addProcessor(name, new KStreamTransform<>(transformerSupplier), this.name); topology.connectProcessorAndStateStores(name, stateStoreNames); return new KStreamImpl<>(topology, name, sourceNodes, true); } KStreamImpl(KStreamBuilder topology, String name, Set<String> sourceNodes,
                       boolean repartitionRequired); @Override KStream<K, V> filter(Predicate<? super K, ? super V> predicate); @Override KStream<K, V> filterNot(final Predicate<? super K, ? super V> predicate); @Override KStream<K1, V> selectKey(final KeyValueMapper<? super K, ? super V, ? extends K1> mapper); @Override KStream<K1, V1> map(KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends K1, ? extends V1>> mapper); @Override KStream<K, V1> mapValues(ValueMapper<? super V, ? extends V1> mapper); @Override void print(); @Override void print(String streamName); @Override void print(Serde<K> keySerde, Serde<V> valSerde); @Override void print(Serde<K> keySerde, Serde<V> valSerde, String streamName); @Override void writeAsText(String filePath); @Override void writeAsText(String filePath, String streamName); @Override void writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde); @Override void writeAsText(String filePath, String streamName, Serde<K> keySerde, Serde<V> valSerde); @Override KStream<K1, V1> flatMap(KeyValueMapper<? super K, ? super V, ? extends Iterable<? extends KeyValue<? extends K1, ? extends V1>>> mapper); @Override KStream<K, V1> flatMapValues(ValueMapper<? super V, ? extends Iterable<? extends V1>> mapper); @Override @SuppressWarnings("unchecked") KStream<K, V>[] branch(Predicate<? super K, ? super 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<? super K, ? super V> partitioner, String topic); @Override void foreach(ForeachAction<? super K, ? super V> action); @Override KStream<K, V> peek(final ForeachAction<? super K, ? super V> action); @Override KStream<K, V> through(Serde<K> keySerde, Serde<V> valSerde, String topic); @Override KStream<K, V> through(StreamPartitioner<? super K, ? super V> partitioner, String topic); @Override KStream<K, V> through(String topic); @Override void to(String topic); @Override void to(StreamPartitioner<? super K, ? super V> partitioner, String topic); @Override void to(Serde<K> keySerde, Serde<V> valSerde, String topic); @SuppressWarnings("unchecked") @Override void to(final Serde<K> keySerde, final Serde<V> valSerde, StreamPartitioner<? super K, ? super V> partitioner, final String topic); @Override KStream<K1, V1> transform(TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames); @Override KStream<K, V1> transformValues(ValueTransformerSupplier<? super V, ? extends V1> valueTransformerSupplier, String... stateStoreNames); @Override void process(final ProcessorSupplier<? super K, ? super V> processorSupplier, String... stateStoreNames); @Override KStream<K, R> join(
            final KStream<K, V1> other,
            final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
            final JoinWindows windows,
            final Serde<K> keySerde,
            final Serde<V> thisValueSerde,
            final Serde<V1> otherValueSerde); @Override KStream<K, R> join(
        final KStream<K, V1> other,
        final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        final JoinWindows windows); @Override KStream<K, R> outerJoin(
        final KStream<K, V1> other,
        final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        final JoinWindows windows,
        final Serde<K> keySerde,
        final Serde<V> thisValueSerde,
        final Serde<V1> otherValueSerde); @Override KStream<K, R> outerJoin(
        final KStream<K, V1> other,
        final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        final JoinWindows windows); @Override KStream<K, R> leftJoin(
        final KStream<K, V1> other,
        final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        final JoinWindows windows,
        final Serde<K> keySerde,
        final Serde<V> thisValSerde,
        final Serde<V1> otherValueSerde); @Override KStream<K, R> leftJoin(
        KStream<K, V1> other,
        ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        JoinWindows windows); @Override KStream<K, R> join(final KTable<K, V1> other, final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KStream<K, R> join(final KTable<K, V1> other,
                                      final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                      final Serde<K> keySerde,
                                      final Serde<V> valueSerde); @Override KStream<K, R> leftJoin(final GlobalKTable<K1, V1> globalTable,
                                              final KeyValueMapper<? super K, ? super V, ? extends K1> keyMapper,
                                              final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KStream<K, V2> join(final GlobalKTable<K1, V1> globalTable,
                                            final KeyValueMapper<? super K, ? super V, ? extends K1> keyMapper,
                                            final ValueJoiner<? super V, ? super V1, ? extends V2> joiner); @Override KStream<K, R> leftJoin(final KTable<K, V1> other, final ValueJoiner<? super V, ? super V1, ? extends R> joiner); KStream<K, R> leftJoin(final KTable<K, V1> other,
                                          final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                          final Serde<K> keySerde,
                                          final Serde<V> valueSerde); @Override KGroupedStream<K1, V> groupBy(KeyValueMapper<? super K, ? super V, K1> selector); @Override KGroupedStream<K1, V> groupBy(KeyValueMapper<? super K, ? super V, K1> selector,
                                              Serde<K1> keySerde,
                                              Serde<V> valSerde); @Override KGroupedStream<K, V> groupByKey(); @Override KGroupedStream<K, V> groupByKey(Serde<K> keySerde,
                                           Serde<V> valSerde); }
KStreamImpl extends AbstractStream<K> implements KStream<K, V> { @Override public <K1, V1> KStream<K1, V1> transform(TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames) { Objects.requireNonNull(transformerSupplier, "transformerSupplier can't be null"); String name = topology.newName(TRANSFORM_NAME); topology.addProcessor(name, new KStreamTransform<>(transformerSupplier), this.name); topology.connectProcessorAndStateStores(name, stateStoreNames); return new KStreamImpl<>(topology, name, sourceNodes, true); } KStreamImpl(KStreamBuilder topology, String name, Set<String> sourceNodes,
                       boolean repartitionRequired); @Override KStream<K, V> filter(Predicate<? super K, ? super V> predicate); @Override KStream<K, V> filterNot(final Predicate<? super K, ? super V> predicate); @Override KStream<K1, V> selectKey(final KeyValueMapper<? super K, ? super V, ? extends K1> mapper); @Override KStream<K1, V1> map(KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends K1, ? extends V1>> mapper); @Override KStream<K, V1> mapValues(ValueMapper<? super V, ? extends V1> mapper); @Override void print(); @Override void print(String streamName); @Override void print(Serde<K> keySerde, Serde<V> valSerde); @Override void print(Serde<K> keySerde, Serde<V> valSerde, String streamName); @Override void writeAsText(String filePath); @Override void writeAsText(String filePath, String streamName); @Override void writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde); @Override void writeAsText(String filePath, String streamName, Serde<K> keySerde, Serde<V> valSerde); @Override KStream<K1, V1> flatMap(KeyValueMapper<? super K, ? super V, ? extends Iterable<? extends KeyValue<? extends K1, ? extends V1>>> mapper); @Override KStream<K, V1> flatMapValues(ValueMapper<? super V, ? extends Iterable<? extends V1>> mapper); @Override @SuppressWarnings("unchecked") KStream<K, V>[] branch(Predicate<? super K, ? super 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<? super K, ? super V> partitioner, String topic); @Override void foreach(ForeachAction<? super K, ? super V> action); @Override KStream<K, V> peek(final ForeachAction<? super K, ? super V> action); @Override KStream<K, V> through(Serde<K> keySerde, Serde<V> valSerde, String topic); @Override KStream<K, V> through(StreamPartitioner<? super K, ? super V> partitioner, String topic); @Override KStream<K, V> through(String topic); @Override void to(String topic); @Override void to(StreamPartitioner<? super K, ? super V> partitioner, String topic); @Override void to(Serde<K> keySerde, Serde<V> valSerde, String topic); @SuppressWarnings("unchecked") @Override void to(final Serde<K> keySerde, final Serde<V> valSerde, StreamPartitioner<? super K, ? super V> partitioner, final String topic); @Override KStream<K1, V1> transform(TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames); @Override KStream<K, V1> transformValues(ValueTransformerSupplier<? super V, ? extends V1> valueTransformerSupplier, String... stateStoreNames); @Override void process(final ProcessorSupplier<? super K, ? super V> processorSupplier, String... stateStoreNames); @Override KStream<K, R> join(
            final KStream<K, V1> other,
            final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
            final JoinWindows windows,
            final Serde<K> keySerde,
            final Serde<V> thisValueSerde,
            final Serde<V1> otherValueSerde); @Override KStream<K, R> join(
        final KStream<K, V1> other,
        final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        final JoinWindows windows); @Override KStream<K, R> outerJoin(
        final KStream<K, V1> other,
        final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        final JoinWindows windows,
        final Serde<K> keySerde,
        final Serde<V> thisValueSerde,
        final Serde<V1> otherValueSerde); @Override KStream<K, R> outerJoin(
        final KStream<K, V1> other,
        final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        final JoinWindows windows); @Override KStream<K, R> leftJoin(
        final KStream<K, V1> other,
        final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        final JoinWindows windows,
        final Serde<K> keySerde,
        final Serde<V> thisValSerde,
        final Serde<V1> otherValueSerde); @Override KStream<K, R> leftJoin(
        KStream<K, V1> other,
        ValueJoiner<? super V, ? super V1, ? extends R> joiner,
        JoinWindows windows); @Override KStream<K, R> join(final KTable<K, V1> other, final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KStream<K, R> join(final KTable<K, V1> other,
                                      final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                      final Serde<K> keySerde,
                                      final Serde<V> valueSerde); @Override KStream<K, R> leftJoin(final GlobalKTable<K1, V1> globalTable,
                                              final KeyValueMapper<? super K, ? super V, ? extends K1> keyMapper,
                                              final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KStream<K, V2> join(final GlobalKTable<K1, V1> globalTable,
                                            final KeyValueMapper<? super K, ? super V, ? extends K1> keyMapper,
                                            final ValueJoiner<? super V, ? super V1, ? extends V2> joiner); @Override KStream<K, R> leftJoin(final KTable<K, V1> other, final ValueJoiner<? super V, ? super V1, ? extends R> joiner); KStream<K, R> leftJoin(final KTable<K, V1> other,
                                          final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                          final Serde<K> keySerde,
                                          final Serde<V> valueSerde); @Override KGroupedStream<K1, V> groupBy(KeyValueMapper<? super K, ? super V, K1> selector); @Override KGroupedStream<K1, V> groupBy(KeyValueMapper<? super K, ? super V, K1> selector,
                                              Serde<K1> keySerde,
                                              Serde<V> valSerde); @Override KGroupedStream<K, V> groupByKey(); @Override KGroupedStream<K, V> groupByKey(Serde<K> keySerde,
                                           Serde<V> valSerde); static final String FILTER_NAME; static final String PEEK_NAME; static final String JOINTHIS_NAME; static final String JOINOTHER_NAME; static final String JOIN_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; static final String REPARTITION_TOPIC_SUFFIX; }