public SchemaKStream<K> selectKey(
      final FormatInfo valueFormat,
      final List<Expression> keyExpression,
      final Optional<KeyFormat> forceInternalKeyFormat,
      final Stacker contextStacker,
      final boolean forceRepartition
  ) {
    final boolean keyFormatChange = forceInternalKeyFormat.isPresent()
        && !forceInternalKeyFormat.get().equals(keyFormat);

    final boolean repartitionNeeded = repartitionNeeded(keyExpression);
    if (!keyFormatChange && !forceRepartition && !repartitionNeeded) {
      return this;
    }

    if ((repartitionNeeded || !forceRepartition) && keyFormat.isWindowed()) {
      throw new KsqlException(
          "Implicit repartitioning of windowed sources is not supported. "
              + "See https://github.com/confluentinc/ksql/issues/4385."
      );
    }

    final ExecutionStep<KStreamHolder<K>> step = ExecutionStepFactory
        .streamSelectKey(contextStacker, sourceStep, keyExpression);

    final KeyFormat newKeyFormat = forceInternalKeyFormat.orElse(keyFormat);
    return new SchemaKStream<>(
        step,
        resolveSchema(step),
        SerdeFeaturesFactory.sanitizeKeyFormat(
            newKeyFormat,
            toSqlTypes(keyExpression),
            true),
        ksqlConfig,
        functionRegistry
    );
  }
@Test
  public void shouldFailIfForceRepartitionWindowedStream() {
    // Given:
    givenInitialKStreamOf(
        "SELECT col0, col2, col3 FROM test1 PARTITION BY col0 EMIT CHANGES;",
        windowedKeyFormat);

    // When:
    final KsqlException e = assertThrows(KsqlException.class, () -> initialSchemaKStream
        .selectKey(
            valueFormat.getFormatInfo(),
            ImmutableList.of(new UnqualifiedColumnReferenceExp(ColumnName.of("COL1"))),
            Optional.empty(),
            childContextStacker,
            true)
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "Implicit repartitioning of windowed sources is not supported. See https://github.com/confluentinc/ksql/issues/4385."));
  }