@Override
    public void writeToTopology(final InternalTopologyBuilder topologyBuilder) {
        final String processorName = processorParameters.processorName();
        final ProcessorSupplier<K, V, ?, ?> processorSupplier = processorParameters.processorSupplier();

        // Stream - Table join (Global or KTable)
        topologyBuilder.addProcessor(processorName, processorSupplier, parentNodeNames());

        // Steam - KTable join only
        if (otherJoinSideNodeName != null) {
            topologyBuilder.connectProcessorAndStateStores(processorName, storeNames);
            bufferName.ifPresent(s -> topologyBuilder.connectProcessorAndStateStores(processorName, s));
            if (gracePeriod != null) {
                for (final String storeName : storeNames) {
                    if (!topologyBuilder.isStoreVersioned(storeName)) {
                        throw new IllegalArgumentException("KTable must be versioned to use a grace period in a stream table join.");
                    }
                    if (gracePeriod.toMillis() > topologyBuilder.getHistoryRetention(storeName)) {
                        throw new IllegalArgumentException("History retention must be at least grace period.");
                    }
                }
            }
        }

    }
Represents a join between a KStream and a KTable or GlobalKTable
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamTableJoinNode.java