@SuppressWarnings("unchecked")
    private void enableVersionedSemantics(final ProcessorParameters<K, ?, ?, ?> processorParameters,
                                          final boolean useVersionedSemantics,
                                          final String parentNodeName) {
        final ProcessorSupplier<K, ?, ?, ?> processorSupplier = processorParameters.processorSupplier();
        if (!(processorSupplier instanceof KTableKTableAbstractJoin)) {
            throw new IllegalStateException("Unexpected processor type for table-table join: " + processorSupplier.getClass().getName());
        }
        final KTableKTableAbstractJoin<K, ?, ?, ?> tableJoin = (KTableKTableAbstractJoin<K, ?, ?, ?>) processorSupplier;

        if (parentNodeName.equals(tableJoin.joinThisParentNodeName())) {
            tableJoin.setUseVersionedSemantics(useVersionedSemantics);
        }
    }
The supplier which provides processor with KTable-KTable join merge functionality.
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java