private KStreamImpl<K, V> repartitionForJoin(final String repartitionName,
                                                 final Serde<K> keySerdeOverride,
                                                 final Serde<V> valueSerdeOverride) {
        final Serde<K> repartitionKeySerde = keySerdeOverride != null ? keySerdeOverride : keySerde;
        final Serde<V> repartitionValueSerde = valueSerdeOverride != null ? valueSerdeOverride : valueSerde;
        final OptimizableRepartitionNodeBuilder<K, V> optimizableRepartitionNodeBuilder =
            OptimizableRepartitionNode.optimizableRepartitionNodeBuilder();
        // we still need to create the repartitioned source each time
        // as it increments the counter which
        // is needed to maintain topology compatibility
        final String repartitionedSourceName = createRepartitionedSource(
            builder,
            repartitionKeySerde,
            repartitionValueSerde,
            repartitionName,
            null,
            optimizableRepartitionNodeBuilder);

        if (repartitionNode == null || !name.equals(repartitionName)) {
            repartitionNode = optimizableRepartitionNodeBuilder.build();
            builder.addGraphNode(graphNode, repartitionNode);
        }

        return new KStreamImpl<>(
            repartitionedSourceName,
            repartitionKeySerde,
            repartitionValueSerde,
            Collections.singleton(repartitionedSourceName),
            false,
            repartitionNode,
            builder);
    }
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