@SuppressWarnings("unchecked")
    private <VO, VR> KStream<K, VR> doStreamTableJoin(final KTable<K, VO> table,
                                                      final ValueJoinerWithKey<? super K, ? super V, ? super VO, ? extends VR> joiner,
                                                      final Joined<K, V, VO> joined,
                                                      final boolean leftJoin) {
        Objects.requireNonNull(table, "table can't be null");
        Objects.requireNonNull(joiner, "joiner can't be null");

        final Set<String> allSourceNodes = ensureCopartitionWith(Collections.singleton((AbstractStream<K, VO>) table));

        final JoinedInternal<K, V, VO> joinedInternal = new JoinedInternal<>(joined);
        final NamedInternal renamed = new NamedInternal(joinedInternal.name());

        final String name = renamed.orElseGenerateWithPrefix(builder, leftJoin ? LEFTJOIN_NAME : JOIN_NAME);

        Optional<String> bufferStoreName = Optional.empty();

        if (joined.gracePeriod() != null) {
            if (!((KTableImpl<K, ?, VO>) table).graphNode.isOutputVersioned().orElse(true)) {
                throw new IllegalArgumentException("KTable must be versioned to use a grace period in a stream table join.");
            }
            bufferStoreName = Optional.of(name + "-Buffer");
            final RocksDBTimeOrderedKeyValueBuffer.Builder<Object, Object> storeBuilder =
                    new RocksDBTimeOrderedKeyValueBuffer.Builder<>(bufferStoreName.get(), joined.gracePeriod(), name);
            builder.addStateStore(new StoreBuilderWrapper(storeBuilder));
        }

        final ProcessorSupplier<K, V, K, ? extends VR> processorSupplier = new KStreamKTableJoin<>(
            ((KTableImpl<K, ?, VO>) table).valueGetterSupplier(),
            joiner,
            leftJoin,
            Optional.ofNullable(joined.gracePeriod()),
            bufferStoreName);

        final ProcessorParameters<K, V, ?, ?> processorParameters = new ProcessorParameters<>(processorSupplier, name);
        final StreamTableJoinNode<K, V> streamTableJoinNode = new StreamTableJoinNode<>(
            name,
            processorParameters,
            ((KTableImpl<K, ?, VO>) table).valueGetterSupplier().storeNames(),
            this.name,
            joined.gracePeriod(),
            bufferStoreName
        );

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

        // do not have serde for joined result
        return new KStreamImpl<>(
            name,
            joined.keySerde() != null ? joined.keySerde() : keySerde,
            null,
            allSourceNodes,
            false,
            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