private <KG, VG, VR> KStream<K, VR> globalTableJoin(final GlobalKTable<KG, VG> globalTable,
                                                        final KeyValueMapper<? super K, ? super V, ? extends KG> keySelector,
                                                        final ValueJoinerWithKey<? super K, ? super V, ? super VG, ? extends VR> joiner,
                                                        final boolean leftJoin,
                                                        final Named named) {
        Objects.requireNonNull(globalTable, "globalTable can't be null");
        Objects.requireNonNull(keySelector, "keySelector can't be null");
        Objects.requireNonNull(joiner, "joiner can't be null");
        Objects.requireNonNull(named, "named can't be null");

        final KTableValueGetterSupplier<KG, VG> valueGetterSupplier =
            ((GlobalKTableImpl<KG, VG>) globalTable).valueGetterSupplier();
        final String name = new NamedInternal(named).orElseGenerateWithPrefix(builder, LEFTJOIN_NAME);
        final ProcessorSupplier<K, V, K, VR> processorSupplier = new KStreamGlobalKTableJoin<>(
            valueGetterSupplier,
            joiner,
            keySelector,
            leftJoin);
        final ProcessorParameters<K, V, ?, ?> processorParameters = new ProcessorParameters<>(processorSupplier, name);
        final StreamTableJoinNode<K, V> streamTableJoinNode =
            new StreamTableJoinNode<>(name, processorParameters, new String[] {}, null, null, Optional.empty());

        if (leftJoin) {
            streamTableJoinNode.labels().add(GraphNode.Label.NULL_KEY_RELAXED_JOIN);
        }
        builder.addGraphNode(graphNode, streamTableJoinNode);

        // do not have serde for joined result
        return new KStreamImpl<>(
            name,
            keySerde,
            null,
            subTopologySourceNodes,
            repartitionRequired,
            streamTableJoinNode,
            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