@Override
    public <KOut, VOut> KStream<KOut, VOut> process(
        final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
        final Named named,
        final String... stateStoreNames
    ) {
        Objects.requireNonNull(processorSupplier, "processorSupplier can't be null");
        Objects.requireNonNull(named, "named can't be null");
        Objects.requireNonNull(stateStoreNames, "stateStoreNames can't be a null array");
        ApiUtils.checkSupplier(processorSupplier);
        for (final String stateStoreName : stateStoreNames) {
            Objects.requireNonNull(stateStoreName, "stateStoreNames can't be null");
        }

        final String name = new NamedInternal(named).name();
        final StatefulProcessorNode<? super K, ? super V> processNode = new StatefulProcessorNode<>(
            name,
            new ProcessorParameters<>(processorSupplier, name),
            stateStoreNames);

        builder.addGraphNode(graphNode, processNode);

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