@SuppressWarnings("deprecation")
  public static <K> KStreamHolder<K> build(
      final KStreamHolder<K> left,
      final KStreamHolder<K> right,
      final StreamStreamJoin<K> join,
      final RuntimeBuildContext buildContext,
      final StreamJoinedFactory streamJoinedFactory) {
    final QueryContext queryContext = join.getProperties().getQueryContext();
    final QueryContext.Stacker stacker = QueryContext.Stacker.of(queryContext);

    final LogicalSchema leftSchema;
    final LogicalSchema rightSchema;
    final Formats rightFormats;
    final Formats leftFormats;
    if (join.getJoinType().equals(RIGHT)) {
      leftFormats = join.getRightInternalFormats();
      rightFormats = join.getLeftInternalFormats();

      leftSchema = right.getSchema();
      rightSchema = left.getSchema();
    } else {
      leftFormats = join.getLeftInternalFormats();
      rightFormats = join.getRightInternalFormats();

      leftSchema = left.getSchema();
      rightSchema = right.getSchema();
    }

    final PhysicalSchema leftPhysicalSchema = PhysicalSchema.from(
        leftSchema,
        leftFormats.getKeyFeatures(),
        leftFormats.getValueFeatures()
    );

    final Serde<GenericRow> leftSerde = buildContext.buildValueSerde(
        leftFormats.getValueFormat(),
        leftPhysicalSchema,
        stacker.push(LEFT_SERDE_CTX).getQueryContext()
    );
    final PhysicalSchema rightPhysicalSchema = PhysicalSchema.from(
        rightSchema,
        rightFormats.getKeyFeatures(),
        rightFormats.getValueFeatures()
    );

    final Serde<GenericRow> rightSerde = buildContext.buildValueSerde(
        rightFormats.getValueFormat(),
        rightPhysicalSchema,
        stacker.push(RIGHT_SERDE_CTX).getQueryContext()
    );
    final Serde<K> keySerde = left.getExecutionKeyFactory().buildKeySerde(
        leftFormats.getKeyFormat(),
        leftPhysicalSchema,
        queryContext
    );
    final StreamJoined<K, GenericRow, GenericRow> joined = streamJoinedFactory.create(
        keySerde,
        leftSerde,
        rightSerde,
        StreamsUtil.buildOpName(queryContext),
        StreamsUtil.buildOpName(queryContext)
    );

    final JoinParams joinParams = JoinParamsFactory
        .create(join.getKeyColName(), leftSchema, rightSchema);

    JoinWindows joinWindows;

    // Grace, as optional, helps to identify if a user specified the GRACE PERIOD syntax in the
    // join window. If specified, then we'll call the new KStreams API ofTimeDifferenceAndGrace()
    // which enables the "spurious" results bugfix with left/outer joins (see KAFKA-10847).
    if (join.getGraceMillis().isPresent()) {
      joinWindows = JoinWindows.ofTimeDifferenceAndGrace(
            join.getBeforeMillis(),
            join.getGraceMillis().get());
    } else {
      joinWindows = JoinWindows.of(join.getBeforeMillis());
    }

    joinWindows = joinWindows.after(join.getAfterMillis());

    final KStream<K, GenericRow> result;
    switch (join.getJoinType()) {
      case LEFT:
        result = left.getStream().leftJoin(
            right.getStream(), joinParams.getJoiner(), joinWindows, joined);
        break;
      case RIGHT:
        result = right.getStream().leftJoin(
            left.getStream(), joinParams.getJoiner(), joinWindows, joined);
        break;
      case OUTER:
        result = left.getStream().outerJoin(
            right.getStream(), joinParams.getJoiner(), joinWindows, joined);
        break;
      case INNER:
        result = left.getStream().join(
            right.getStream(), joinParams.getJoiner(), joinWindows, joined);
        break;
      default:
        throw new IllegalStateException("invalid join type");
    }
    return left.withStream(result, joinParams.getSchema());
  }
@Test
  public void shouldReturnCorrectLegacySchema() {
    // Given:
    givenInnerJoin(L_KEY);

    join = new StreamStreamJoin<>(
        new ExecutionStepPropertiesV1(CTX),
        JoinType.INNER,
        ColumnName.of(LEGACY_KEY_COL),
        LEFT_FMT,
        RIGHT_FMT,
        left,
        right,
        BEFORE,
        AFTER,
        Optional.empty()
    );

    // When:
    final KStreamHolder<Struct> result = join.build(planBuilder, planInfo);

    // Then:
    assertThat(
        result.getSchema(),
        is(JoinParamsFactory.create(ROWKEY_NAME, LEFT_SCHEMA, RIGHT_SCHEMA).getSchema())
    );
  }