@Override
    public <VO, VR> KStream<K, VR> leftJoin(final KTable<K, VO> table,
                                            final ValueJoinerWithKey<? super K, ? super V, ? super VO, ? extends VR> joiner,
                                            final Joined<K, V, VO> joined) {
        Objects.requireNonNull(table, "table can't be null");
        Objects.requireNonNull(joiner, "joiner can't be null");
        Objects.requireNonNull(joined, "joined can't be null");
        final JoinedInternal<K, V, VO> joinedInternal = new JoinedInternal<>(joined);
        final String name = joinedInternal.name();

        if (repartitionRequired) {
            final KStreamImpl<K, V> thisStreamRepartitioned = repartitionForJoin(
                    name != null ? name : this.name,
                    joined.keySerde(),
                    joined.valueSerde()
            );
            return thisStreamRepartitioned.doStreamTableJoin(table, joiner, joined, true);
        } else {
            return doStreamTableJoin(table, joiner, joined, true);
        }
    }
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