private <VR> KStream<K, VR> doTransformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerWithKeySupplier,
                                                  final NamedInternal named,
                                                  final String... stateStoreNames) {
        Objects.requireNonNull(stateStoreNames, "stateStoreNames can't be a null array");
        for (final String stateStoreName : stateStoreNames) {
            Objects.requireNonNull(stateStoreName, "stateStoreNames can't contain `null` as store name");
        }
        ApiUtils.checkSupplier(valueTransformerWithKeySupplier);

        final String name = named.orElseGenerateWithPrefix(builder, TRANSFORMVALUES_NAME);
        final StatefulProcessorNode<? super K, ? super V> transformNode = new StatefulProcessorNode<>(
            name,
            new ProcessorParameters<>(new KStreamTransformValues<>(valueTransformerWithKeySupplier), name),
            stateStoreNames);
        transformNode.setValueChangingOperation(true);

        builder.addGraphNode(graphNode, transformNode);

        // cannot inherit value serde
        return new KStreamImpl<>(
            name,
            keySerde,
            null,
            subTopologySourceNodes,
            repartitionRequired,
            transformNode,
            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