@SuppressWarnings("unchecked")
  @Override
  public <T extends Statement> ConfiguredStatement<T> inject(
      final ConfiguredStatement<T> statement
  ) {
    if (statement.getStatement() instanceof CreateSource) {
      return handleCreateSource((ConfiguredStatement<CreateSource>) statement);
    }

    return statement;
  }
@Test
  public void shouldThrowIfMissingDefaultValueFormatConfig() {
    // Given
    givenConfig(ImmutableMap.of());
    givenSourceProps(ImmutableMap.of(
        "KEY_FORMAT", new StringLiteral("KAFKA")
    ));

    // Expect / When
    final Exception e = assertThrows(
        KsqlStatementException.class,
        () -> injector.inject(csStatement)
    );

    // Then
    assertThat(e.getMessage(), containsString(
        "Statement is missing the 'VALUE_FORMAT' property from the WITH clause."));
    assertThat(e.getMessage(), containsString("Either provide one or set a default via the '"
        + KsqlConfig.KSQL_DEFAULT_VALUE_FORMAT_CONFIG + "' config."));
  }