@Override
    @Deprecated
    public <K1, V1> KStream<K1, V1> flatTransform(final TransformerSupplier<? super K, ? super V, Iterable<KeyValue<K1, V1>>> transformerSupplier,
                                                  final Named named,
                                                  final String... stateStoreNames) {
        Objects.requireNonNull(transformerSupplier, "transformerSupplier can't be null");
        Objects.requireNonNull(named, "named can't be null");
        Objects.requireNonNull(stateStoreNames, "stateStoreNames can't be a null array");
        ApiUtils.checkSupplier(transformerSupplier);
        for (final String stateStoreName : stateStoreNames) {
            Objects.requireNonNull(stateStoreName, "stateStoreNames can't contain `null` as store name");
        }

        final String name = new NamedInternal(named).name();
        final StatefulProcessorNode<? super K, ? super V> transformNode = new StatefulProcessorNode<>(
            name,
            new ProcessorParameters<>(new KStreamFlatTransform<>(transformerSupplier), name),
            stateStoreNames);
        transformNode.keyChangingOperation(true);

        builder.addGraphNode(graphNode, transformNode);

        // cannot inherit key and value serde
        return new KStreamImpl<>(
            name,
            null,
            null,
            subTopologySourceNodes,
            true,
            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