static <K1, V1, RN extends BaseRepartitionNode<K1, V1>> String createRepartitionedSource(final InternalStreamsBuilder builder,
                                                                                             final Serde<K1> keySerde,
                                                                                             final Serde<V1> valueSerde,
                                                                                             final String repartitionTopicNamePrefix,
                                                                                             final StreamPartitioner<K1, V1> streamPartitioner,
                                                                                             final BaseRepartitionNodeBuilder<K1, V1, RN> baseRepartitionNodeBuilder) {

        final String repartitionTopicName = repartitionTopicNamePrefix.endsWith(REPARTITION_TOPIC_SUFFIX) ?
            repartitionTopicNamePrefix :
            repartitionTopicNamePrefix + REPARTITION_TOPIC_SUFFIX;

        // Always need to generate the names to burn index counter for compatibility
        final String genSinkName = builder.newProcessorName(SINK_NAME);
        final String genNullKeyFilterProcessorName = builder.newProcessorName(FILTER_NAME);
        final String genSourceName = builder.newProcessorName(SOURCE_NAME);

        final String sinkName;
        final String sourceName;
        final String nullKeyFilterProcessorName;

        if (repartitionTopicNamePrefix.matches("KSTREAM.*-[0-9]{10}")) {
            sinkName = genSinkName;
            sourceName = genSourceName;
            nullKeyFilterProcessorName = genNullKeyFilterProcessorName;
        } else {
            sinkName = repartitionTopicName + "-sink";
            sourceName = repartitionTopicName + "-source";
            nullKeyFilterProcessorName = repartitionTopicName + "-filter";
        }

        final ProcessorParameters<K1, V1, ?, ?> processorParameters = new ProcessorParameters<>(
            new KStreamFilter<>((k, v) -> true, false),
            nullKeyFilterProcessorName
        );

        baseRepartitionNodeBuilder.withKeySerde(keySerde)
                                  .withValueSerde(valueSerde)
                                  .withSourceName(sourceName)
                                  .withRepartitionTopic(repartitionTopicName)
                                  .withSinkName(sinkName)
                                  .withProcessorParameters(processorParameters)
                                  .withStreamPartitioner(streamPartitioner)
                                  // reusing the source name for the graph node name
                                  // adding explicit variable as it simplifies logic
                                  .withNodeName(sourceName);

        return sourceName;
    }
Repartition a stream. This is required on join operations occurring after
an operation that changes the key, i.e, selectKey, map(..), flatMap(..).
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java