public static JoinParams create(
      final ColumnName keyColName,
      final LogicalSchema leftSchema,
      final LogicalSchema rightSchema
  ) {
    final boolean appendKey = neitherContain(keyColName, leftSchema, rightSchema);
    return new JoinParams(
        new KsqlValueJoiner(leftSchema.value().size(), rightSchema.value().size(), appendKey ? 1 : 0
        ),
        createSchema(keyColName, leftSchema, rightSchema)
    );
  }
@Test
  public void shouldThrowOnKeyTypeMismatch() {
    // Given:
    final LogicalSchema intKeySchema = LogicalSchema.builder()
        .keyColumn(ColumnName.of("BOB"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("BLUE"), SqlTypes.STRING)
        .valueColumn(ColumnName.of("GREEN"), SqlTypes.INTEGER)
        .build()
        .withPseudoAndKeyColsInValue(false);

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> JoinParamsFactory.create(ColumnName.of("BOB"), intKeySchema, RIGHT_SCHEMA)
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "Invalid join. Key types differ: INTEGER vs STRING"));
  }