public static Format of(final FormatInfo formatInfo) {
    final Format format = fromName(formatInfo.getFormat().toUpperCase());
    format.validateProperties(formatInfo.getProperties());
    return format;
  }
@Test
  public void shouldThrowWhenCreatingFromUnsupportedProperty() {
    // Given:
    final FormatInfo format = FormatInfo.of("JSON", ImmutableMap.of("KEY_SCHEMA_ID", "1"));
    final FormatInfo kafkaFormat = FormatInfo.of("KAFKA", ImmutableMap.of("VALUE_SCHEMA_ID", "1"));
    final FormatInfo delimitedFormat = FormatInfo.of("delimited",
        ImmutableMap.of("KEY_SCHEMA_ID", "123"));

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> FormatFactory.of(format)
    );
    final Exception kafkaException = assertThrows(
        KsqlException.class,
        () -> FormatFactory.of(kafkaFormat)
    );
    final Exception delimitedException = assertThrows(
        KsqlException.class,
        () -> FormatFactory.of(delimitedFormat)
    );
    // Then:
    assertThat(e.getMessage(),
        containsString("JSON does not support the following configs: [KEY_SCHEMA_ID]"));
    assertThat(kafkaException.getMessage(),
        containsString("KAFKA does not support the following configs: [VALUE_SCHEMA_ID]"));
    assertThat(delimitedException.getMessage(),
        containsString("DELIMITED does not support the following configs: [KEY_SCHEMA_ID]"));
  }