KIP-221 / Add KStream#repartition operation

streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java

@@ -18,6 +18,7 @@
 
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsBuilder;
@@ -814,6 +815,8 @@ <VR> KStream<K, VR> flatMapValues(final ValueMapperWithKey<? super K, ? super V,
      *
      * @param topic the topic name
      * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
+     * @see #repartition()
+     * @see #repartition(Repartitioned)
      */
     KStream<K, V> through(final String topic);
 
@@ -832,10 +835,52 @@ <VR> KStream<K, VR> flatMapValues(final ValueMapperWithKey<? super K, ? super V,
      * @param topic     the topic name
      * @param produced  the options to use when producing to the topic
      * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
+     * @see #repartition()
+     * @see #repartition(Repartitioned)
      */
     KStream<K, V> through(final String topic,
                           final Produced<K, V> produced);
 
+    /**
+     * Materialize this stream to an auto-generated repartition topic and create a new {@code KStream}
+     * from the auto-generated topic using default serializers, deserializers, and producer's {@link DefaultPartitioner}.
+     * The number of partitions is determined based on the upstream topics partition numbers.
+     * <p>
+     * This operation is similar to {@link #through(String)}, however, Kafka Streams manages the used topic automatically.
+     * The created topic is considered as an internal topic and is meant to be used only by the current Kafka Streams instance.
+     * Similar to auto-repartitioning, the topic will be created with infinite retention time and data will be automatically purged by Kafka Streams.
+     * The topic will be named as "${applicationId}-&lt;name&gt;-repartition", where "applicationId" is user-specified in
+     * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
+     * "&lt;name&gt;" is an internally generated name, and "-repartition" is a fixed suffix.
+     *
+     * @return {@code KStream} that contains the exact same repartitioned records as this {@code KStream}.
+     * @see #through(String)
+     * @see #through(String, Produced)
+     */
+    KStream<K, V> repartition();
+
+    /**
+     * Materialize this stream to an auto-generated repartition topic and create a new {@code KStream}
+     * from the auto-generated topic using {@link Serde key serde}, {@link Serde value serde}, {@link StreamPartitioner},
+     * number of partitions, and topic name part as defined by {@link Repartitioned}.
+     * <p>
+     * This operation is similar to {@link #through(String)}, however, Kafka Streams manages the used topic automatically.
+     * The created topic is considered as an internal topic and is meant to be used only by the current Kafka Streams instance.
+     * Similar to auto-repartitioning, the topic will be created with infinite retention time and data will be automatically purged by Kafka Streams.
+     * The topic will be named as "${applicationId}-&lt;name&gt;-repartition", where "applicationId" is user-specified in
+     * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
+     * "&lt;name&gt;" is either provided via {@link Repartitioned#as(String)} or an internally
+     * generated name, and "-repartition" is a fixed suffix.
+     *
+     * @param repartitioned the {@link Repartitioned} instance used to specify {@link Serdes},
+     *                      {@link StreamPartitioner} which determines how records are distributed among partitions of the topic,
+     *                      part of the topic name, and number of partitions for a repartition topic.
+     * @return a {@code KStream} that contains the exact same repartitioned records as this {@code KStream}.
+     * @see #through(String)
+     * @see #through(String, Produced)
+     */
+    KStream<K, V> repartition(final Repartitioned<K, V> repartitioned);
+
     /**
      * Materialize this stream to a topic using default serializers specified in the config and producer's
      * {@link DefaultPartitioner}.
@@ -1172,7 +1217,7 @@ <KR> KGroupedStream<KR, V> groupBy(final KeyValueMapper<? super K, ? super V, KR
      * records to it, and rereading all records from it, such that the resulting {@link KGroupedStream} is partitioned
      * correctly on its key.
      *
-     * @param  grouped  the {@link Grouped} instance used to specify {@link org.apache.kafka.common.serialization.Serdes}
+     * @param  grouped  the {@link Grouped} instance used to specify {@link Serdes}
      *                  and part of the name for a repartition topic if repartitioning is required.
      * @return a {@link KGroupedStream} that contains the grouped records of the original {@code KStream}
      * @see #groupBy(KeyValueMapper)

streams/src/main/java/org/apache/kafka/streams/kstream/Repartitioned.java

@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.kstream;
+
+import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.kstream.internals.WindowedSerializer;
+import org.apache.kafka.streams.kstream.internals.WindowedStreamPartitioner;
+import org.apache.kafka.streams.processor.StreamPartitioner;
+
+/**
+ * This class is used to provide the optional parameters for internal repartition topics.
+ *
+ * @param <K> key type
+ * @param <V> value type
+ * @see KStream#repartition()
+ * @see KStream#repartition(Repartitioned)
+ */
+public class Repartitioned<K, V> implements NamedOperation<Repartitioned<K, V>> {
+
+    protected final String name;
+    protected final Serde<K> keySerde;
+    protected final Serde<V> valueSerde;
+    protected final Integer numberOfPartitions;
+    protected final StreamPartitioner<K, V> partitioner;
+
+    private Repartitioned(final String name,
+                          final Serde<K> keySerde,
+                          final Serde<V> valueSerde,
+                          final Integer numberOfPartitions,
+                          final StreamPartitioner<K, V> partitioner) {
+        this.name = name;
+        this.keySerde = keySerde;
+        this.valueSerde = valueSerde;
+        this.numberOfPartitions = numberOfPartitions;
+        this.partitioner = partitioner;
+    }
+
+    protected Repartitioned(final Repartitioned<K, V> repartitioned) {
+        this(
+            repartitioned.name,
+            repartitioned.keySerde,
+            repartitioned.valueSerde,
+            repartitioned.numberOfPartitions,
+            repartitioned.partitioner
+        );
+    }
+
+    /**
+     * Create a {@code Repartitioned} instance with the provided name used as part of the repartition topic.
+     *
+     * @param name the name used as a processor named and part of the repartition topic name.
+     * @param <K>  key type
+     * @param <V>  value type
+     * @return A new {@code Repartitioned} instance configured with processor name and repartition topic name
+     * @see KStream#repartition(Repartitioned)
+     */
+    public static <K, V> Repartitioned<K, V> as(final String name) {
+        return new Repartitioned<>(name, null, null, null, null);
+    }
+
+    /**
+     * Create a {@code Repartitioned} instance with provided key serde and value serde.
+     *
+     * @param keySerde   Serde to use for serializing the key
+     * @param valueSerde Serde to use for serializing the value
+     * @param <K>        key type
+     * @param <V>        value type
+     * @return A new {@code Repartitioned} instance configured with key serde and value serde
+     * @see KStream#repartition(Repartitioned)
+     */
+    public static <K, V> Repartitioned<K, V> with(final Serde<K> keySerde,
+                                                  final Serde<V> valueSerde) {
+        return new Repartitioned<>(null, keySerde, valueSerde, null, null);
+    }
+
+    /**
+     * Create a {@code Repartitioned} instance with provided partitioner.
+     *
+     * @param partitioner the function used to determine how records are distributed among partitions of the topic,
+     *                    if not specified and the key serde provides a {@link WindowedSerializer} for the key
+     *                    {@link WindowedStreamPartitioner} will be used—otherwise {@link DefaultPartitioner} will be used
+     * @param <K>         key type
+     * @param <V>         value type
+     * @return A new {@code Repartitioned} instance configured with partitioner
+     * @see KStream#repartition(Repartitioned)
+     */
+    public static <K, V> Repartitioned<K, V> streamPartitioner(final StreamPartitioner<K, V> partitioner) {
+        return new Repartitioned<>(null, null, null, null, partitioner);
+    }
+
+    /**
+     * Create a {@code Repartitioned} instance with provided number of partitions for repartition topic.
+     *
+     * @param numberOfPartitions number of partitions used when creating repartition topic
+     * @param <K>                key type
+     * @param <V>                value type
+     * @return A new {@code Repartitioned} instance configured number of partitions
+     * @see KStream#repartition(Repartitioned)
+     */
+    public static <K, V> Repartitioned<K, V> numberOfPartitions(final int numberOfPartitions) {
+        return new Repartitioned<>(null, null, null, numberOfPartitions, null);
+    }
+
+    /**
+     * Create a new instance of {@code Repartitioned} with the provided name used as part of repartition topic and processor name.
+     *
+     * @param name the name used for the processor name and as part of the repartition topic
+     * @return a new {@code Repartitioned} instance configured with the name
+     */
+    @Override
+    public Repartitioned<K, V> withName(final String name) {
+        return new Repartitioned<>(name, keySerde, valueSerde, numberOfPartitions, partitioner);
+    }
+
+    /**
+     * Create a new instance of {@code Repartitioned} with the provided number of partitions for repartition topic.
+     *
+     * @param numberOfPartitions the name used for the processor name and as part of the repartition topic name
+     * @return a new {@code Repartitioned} instance configured with the number of partitions
+     */
+    public Repartitioned<K, V> withNumberOfPartitions(final int numberOfPartitions) {
+        return new Repartitioned<>(name, keySerde, valueSerde, numberOfPartitions, partitioner);
+    }
+
+    /**
+     * Create a new instance of {@code Repartitioned} with the provided key serde.
+     *
+     * @param keySerde Serde to use for serializing the key
+     * @return a new {@code Repartitioned} instance configured with the key serde
+     */
+    public Repartitioned<K, V> withKeySerde(final Serde<K> keySerde) {
+        return new Repartitioned<>(name, keySerde, valueSerde, numberOfPartitions, partitioner);
+    }
+
+    /**
+     * Create a new instance of {@code Repartitioned} with the provided value serde.
+     *
+     * @param valueSerde Serde to use for serializing the value
+     * @return a new {@code Repartitioned} instance configured with the value serde
+     */
+    public Repartitioned<K, V> withValueSerde(final Serde<V> valueSerde) {
+        return new Repartitioned<>(name, keySerde, valueSerde, numberOfPartitions, partitioner);
+    }
+
+    /**
+     * Create a new instance of {@code Repartitioned} with the provided partitioner.
+     *
+     * @param partitioner the function used to determine how records are distributed among partitions of the topic,
+     *                    if not specified and the key serde provides a {@link WindowedSerializer} for the key
+     *                    {@link WindowedStreamPartitioner} will be used—otherwise {@link DefaultPartitioner} wil be used
+     * @return a new {@code Repartitioned} instance configured with provided partitioner
+     */
+    public Repartitioned<K, V> withStreamPartitioner(final StreamPartitioner<K, V> partitioner) {
+        return new Repartitioned<>(name, keySerde, valueSerde, numberOfPartitions, partitioner);
+    }
+}

streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedStreamAggregateBuilder.java

@@ -177,6 +177,7 @@ private <VIn> void createRepartitionSource(final String repartitionTopicNamePref
                 keySerde,
                 (Serde<VIn>) valueSerde,
                 repartitionTopicNamePrefix,
+                null,
                 (OptimizableRepartitionNodeBuilder<K, VIn>) optimizableRepartitionNodeBuilder);
 
     }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/GroupedStreamAggregateBuilder.java

@@ -126,6 +126,7 @@ private String createRepartitionSource(final String repartitionTopicNamePrefix,
                                                      keySerde,
                                                      valueSerde,
                                                      repartitionTopicNamePrefix,
+                                                     null,
                                                      optimizableRepartitionNodeBuilder);
 
     }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java

@@ -435,6 +435,7 @@ private <K, V> OptimizableRepartitionNode<K, V> createRepartitionNode(final Stri
                 keySerde,
                 valueSerde,
                 repartitionTopicName,
+                null,
                 repartitionNodeBuilder
         );