public KsqlConfig cloneWithPropertyOverwrite(final Map<String, ?> props) {
    final Map<String, Object> cloneProps = new HashMap<>(originals());
    cloneProps.putAll(props);
    final Map<String, ConfigValue> streamConfigProps =
        buildStreamingConfig(getKsqlStreamConfigProps(), props);

    return new KsqlConfig(ConfigGeneration.CURRENT, cloneProps, streamConfigProps);
  }
@Test
  public void shouldHaveCorrectOriginalsAfterCloneWithOverwrite() {
    // Given:
    final KsqlConfig initial = new KsqlConfig(ImmutableMap.of(
        KsqlConfig.KSQL_SERVICE_ID_CONFIG, "original-id",
        KsqlConfig.KSQL_WRAP_SINGLE_VALUES, "true"
    ));

    // When:
    final KsqlConfig cloned = initial.cloneWithPropertyOverwrite(ImmutableMap.of(
        KsqlConfig.KSQL_SERVICE_ID_CONFIG, "overridden-id",
        KsqlConfig.KSQL_PERSISTENT_QUERY_NAME_PREFIX_CONFIG, "bob"
    ));

    // Then:
    assertThat(cloned.originals(), is(ImmutableMap.of(
        KsqlConfig.KSQL_SERVICE_ID_CONFIG, "overridden-id",
        KsqlConfig.KSQL_WRAP_SINGLE_VALUES, "true",
        KsqlConfig.KSQL_PERSISTENT_QUERY_NAME_PREFIX_CONFIG, "bob"
    )));
  }