@VisibleForTesting
  static void validateDefaultTopicFormats(final KsqlConfig config) {
    validateTopicFormat(config, KsqlConfig.KSQL_DEFAULT_KEY_FORMAT_CONFIG, "key");
    validateTopicFormat(config, KsqlConfig.KSQL_DEFAULT_VALUE_FORMAT_CONFIG, "value");
  }
@Test
  public void shouldFailOnInvalidDefaultKeyFormat() {
    // Given:
    final KsqlConfig config = configWith(ImmutableMap.of(
        KsqlConfig.KSQL_DEFAULT_KEY_FORMAT_CONFIG, "bad"
    ));

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> KsqlServerMain.validateDefaultTopicFormats(config)
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "Invalid value for config '" + KsqlConfig.KSQL_DEFAULT_KEY_FORMAT_CONFIG + "': bad"));
  }