@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 shouldNotPerformStreamToTableJoinIfJoinWindowIsSpecified() {
    // Given:
    when(left.getNodeOutputType()).thenReturn(DataSourceType.KSTREAM);
    when(right.getNodeOutputType()).thenReturn(DataSourceType.KTABLE);

    final WithinExpression withinExpression = new WithinExpression(10, TimeUnit.SECONDS);

    final JoinNode joinNode =
        new JoinNode(nodeId, OUTER, joinKey,           true, left, right,
            Optional.of(withinExpression),"KAFKA");

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> joinNode.buildStream(planBuildContext)
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "A window definition was provided for a Stream-Table join."));
  }