@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 shouldInjectMissingKeyFormat() {
    // Given
    givenConfig(ImmutableMap.of(
        KsqlConfig.KSQL_DEFAULT_KEY_FORMAT_CONFIG, "KAFKA"
    ));
    givenSourceProps(ImmutableMap.of(
        "VALUE_FORMAT", new StringLiteral("JSON")
    ));

    // When
    final ConfiguredStatement<?> result = injector.inject(csStatement);

    // Then
    assertThat(result.getMaskedStatementText(), containsString("KEY_FORMAT='KAFKA'"));
  }