@Override
  public SchemaKStream<?> buildStream(final PlanBuildContext buildContext) {

    if (!joinKey.isForeignKey()) {
      ensureMatchingPartitionCounts(buildContext.getServiceContext().getTopicClient());
    }

    final JoinerFactory joinerFactory = new JoinerFactory(
        buildContext,
        this,
        buildContext.buildNodeContext(getId().toString()));

    return joinerFactory.getJoiner(left.getNodeOutputType(), right.getNodeOutputType()).join();
  }
@Test
  public void shouldHandleJoinIfTableHasNoKeyAndJoinFieldIsRowKey() {
    // Given:
    setupStream(left, leftSchemaKStream);
    setupTable(right, rightSchemaKTable);

    final JoinNode joinNode = new JoinNode(nodeId, LEFT, joinKey,        true, left,
        right, empty(), "KAFKA");

    // When:
    joinNode.buildStream(planBuildContext);

    // Then:
    verify(leftSchemaKStream).leftJoin(
        rightSchemaKTable,
        SYNTH_KEY,
        VALUE_FORMAT.getFormatInfo(),
        CONTEXT_STACKER
    );
  }