/**
 * This class is used to provide the optional parameters for internal repartitioned topics when using
 * using:
 * - {@link KStream#repartition(Repartitioned)}
 * - {@link KStream#repartition(KeyValueMapper, Repartitioned)}
 * - {@link KStream#groupByKey(Repartitioned)}
 * - {@link KStream#groupBy(KeyValueMapper, Repartitioned)}
 *
 * @param @param <K> key type
 * @param <V> value type
 */
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<? super K, ? super V> partitioner;

	private Repartitioned(String name,
	                      Serde<K> keySerde,
	                      Serde<V> valueSerde,
	                      Integer numberOfPartitions,
	                      StreamPartitioner<? super K, ? super V> partitioner) {
		this.name = name;
		this.keySerde = keySerde;
		this.valueSerde = valueSerde;
		this.numberOfPartitions = numberOfPartitions;
		this.partitioner = partitioner;
	}

	/**
	 * Create a {@link Repartitioned} instance with the provided name used as part of the repartition topic if required.
	 *
	 * @param name the name used as a processor named and part of the repartition topic name if required.
	 * @param <K>  key type
	 * @param <V>  value type
	 * @return A new {@link Repartitioned} instance configured with processor name and repartition topic name
	 * @see KStream#repartition(Repartitioned)
	 * @see KStream#repartition(KeyValueMapper, Repartitioned)
	 * @see KStream#groupByKey(Repartitioned)
	 * @see KStream#groupBy(KeyValueMapper, Repartitioned)
	 */
	public static <K, V> Repartitioned<K, V> as(final String name) {
		return new Repartitioned<>(name, null, null, null, null);
	}

	/**
	 * Create a {@link 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 {@link Repartitioned} instance configured with key serde and value serde
	 * @see KStream#repartition(Repartitioned)
	 * @see KStream#repartition(KeyValueMapper, Repartitioned)
	 * @see KStream#groupByKey(Repartitioned)
	 * @see KStream#groupBy(KeyValueMapper, 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 {@link Repartitioned} instance with provided name and number of partitions usedpartitioner.
	 *
	 * as@param part ofpartitioner the repartitionfunction topicused configurationto if required.
	 *
	 * @param <K>    determine how records are distributed among partitions of the topic,
	 *            key type
	 * @param <V>    if not specified and the key serde provides a {@link WindowedSerializer} for valuethe typekey
	 * @param  name               the name used{@link asWindowedStreamPartitioner} partwill ofbe theused—otherwise repartition{@link topicDefaultPartitioner} namewill ifbe requiredused
	 * @param numberOfPartitions<K> number of partitions used for creating repartition topic ifkey requiredtype
	 * @return@param A<V> new instance of {@link Repartitioned} with configured name and number of partitionsvalue type
	 * @see KStream#repartition(Repartitioned)
	 * @see KStream#repartition(KeyValueMapper, Repartitioned) @return A new {@link Repartitioned} instance configured with partitioner
	 * @see KStream#groupByKeyKStream#repartition(Repartitioned)
	 * @see KStream#groupByKStream#repartition(KeyValueMapper, Repartitioned)
	 */
	public static <K, V> Repartitioned<K, V> withstreamPartitioner(final StreamPartitioner<? Stringsuper nameK,
	 ? super V> partitioner) {
		return                                         final int numberOfPartitions) {
		return new Repartitioned<>(namenew Repartitioned<>(null, null, null, numberOfPartitionsnull, nullpartitioner);
	}

	/**
	 * Create a {@link Repartitioned} instance with provided partitioner number of partitions for repartition topic if required.
	 *
	 * @param partitionernumberOfPartitions thenumber functionof partitions used towhen creating determinerepartition howtopic recordsif arerequired
	 distributed* among@param partitions<K> of the topic,
	 *            key type
	 * @param <V>    if not specified and the key serde provides a {@link WindowedSerializer} for thevalue keytype
	 * @return A new {@link Repartitioned} instance configured number of partitions
	 * @see KStream#repartition(Repartitioned)
	 * @see KStream#repartition(KeyValueMapper, Repartitioned)
	 */
	public static <K, V> {@link WindowedStreamPartitioner} will be used—otherwise {@link DefaultPartitioner} will be usedRepartitioned<K, V> numberOfPartitions(final int numberOfPartitions) {
		return new Repartitioned<>(null, null, null, numberOfPartitions, null);
	}

	/**
	 * @paramCreate <K>a new instance of {@link Repartitioned} with the provided keyname type
	used *as @parampart <V>of repartition topic and      value typeprocessor name.
	 * @returnNote that AKafka newStreams {@linkcreates Repartitioned}repartition instancetopic configuredonly withif partitionerrequired.
	 * @see KStream#repartition(Repartitioned)
	 * @see KStream#repartition(KeyValueMapper, Repartitioned)
	 * @see KStream#groupByKey(Repartitioned)
	 * @see KStream#groupBy(KeyValueMapper, Repartitioned)@param name the name used for the processor name and as part of the repartition topic name if required
	 * @return a new {@link Repartitioned} instance configured with the name
	 */
	public static <K, V>@Override
	public Repartitioned<K, V> streamPartitionerwithName(StreamPartitioner<?final super K, ? super V> partitionerString name) {
		return new Repartitioned<>(nullname, nullkeySerde, nullvalueSerde, nullnumberOfPartitions, partitioner);
	}

	/**
	 * Create a new instance of {@link Repartitioned} instancewith withthe provided number of partitions for repartition topic if required.
	 *
	 *Note @paramthat numberOfPartitionsKafka number of partitions used when creating Streams creates repartition topic only if required.
	 *
	 * @param <K>numberOfPartitions the name used for the processor name and as part of the repartition topic name keyif typerequired
	 * @param@return <V>a new {@link Repartitioned} instance configured with the number of partitions
	 */
	public Repartitioned<K, V> withNumberOfPartitions(final int  value typenumberOfPartitions) {
		return new Repartitioned<>(name, keySerde, valueSerde, numberOfPartitions, partitioner);
	}

	/**
	 * @returnCreate Aa new instance of {@link Repartitioned} instancewith configuredthe numberprovided ofkey partitionsserde.
	 * @see KStream#repartition(Repartitioned)
	 * @see KStream#repartition(KeyValueMapper, Repartitioned)@param keySerde Serde to use for serializing the key
	 * @see KStream#groupByKey(Repartitioned)
	 * @see KStream#groupBy(KeyValueMapper, Repartitioned) @return a new {@link Repartitioned} instance configured with the key serde
	 */
	public static <K, V> Repartitioned<K, V> numberOfPartitionswithKeySerde(intfinal Serde<K> numberOfPartitionskeySerde) {
		return new Repartitioned<>(nullname, nullkeySerde, nullvalueSerde, numberOfPartitions, nullpartitioner);
	}

	/**
	 * Create a new instance of {@link Repartitioned} with the provided name used as part of repartition topic and processor name.value serde.
	 * Note that Kafka Streams creates repartition topic only if required.
	 *
	 * @param namevalueSerde theSerde nameto useduse for the processor name and as part of the repartition topic name if requiredserializing the value
	 * @return a new {@link Repartitioned} instance configured with the value nameserde
	 */
	@Override
	public Repartitioned<K, V> withNamewithValueSerde(final StringSerde<V> namevalueSerde) {
		return new Repartitioned<>(name, keySerde, valueSerde, numberOfPartitions, partitioner);
	}

	/**
	 * Create a new instance of {@link Repartitioned} with the provided number of partitions for repartition topic.
	 * Note that Kafka Streams creates repartition topic only if required.
	 *
	 * @param numberOfPartitions the name used for the processor name and as part of the repartition topic name if required
	 * @return a new {@link 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 {@link Repartitioned} with the provided key serde.
	 *
	 * @param keySerde Serde to use for serializing the keypartitioner.
	 *
	 * @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 {@link Repartitioned} instance configured with theprovided key serdepartitioner
	 */
	public Repartitioned<K, V> withKeySerdewithStreamPartitioner(final Serde<K> keySerde StreamPartitioner<? super K, ? super V> partitioner) {
		return new Repartitioned<>(name, keySerde, valueSerde, numberOfPartitions, partitioner);
	}

	/**
	 * Create a new instance of {@link Repartitioned} with the provided value serde.
	 *
	 * @param valueSerde Serde to use for serializing the value
	 * @return a new {@link Repartitioned} instance configured with the value serde
	 */
	public Repartitioned<K, V> withValueSerde(final Serde<V> valueSerde) {
		return new Repartitioned<>(name, keySerde, valueSerde, numberOfPartitions, partitioner);
	}}

and topic name part as defined by {@link Repartitioned}.
	 *
	 * @param repartitioned the {@link Repartitioned} instance used to specify {@link org.apache.kafka.common.serialization.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, if repartitioning is required.
	 * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
	 * @see #repartition()
	 * @see #repartition(KeyValueMapper, Repartitioned)
	 */
	KStream<K, V> repartition(final Repartitioned<K, V> repartitioned);
}

If repartitioned topic is created, it will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in
	 * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "<name>"
     * is either provided via {@link org.apache.kafka.streams.kstream.Repartitioned#as(String)} or generated internally, and "-repartition" is a fixed suffix.
	 * <p>
	 * You can retrieve all generated internal topic names via {@link Topology#describe()}.
	 *
	 * @param repartitioned the {@link Repartitioned} instance used to specify {@link org.apache.kafka.common.serialization.Serdes},
	 *                      {@link org.apache.kafka.streams.processor.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, if repartitioning is required.
	 * @return a {@link KGroupedStream} that contains the grouped records of the original {@code KTable}
	 */
	KGroupedStream<K, V> groupBy(final Repartitioned<K, V> repartitioned);
}