@Override
  public <T extends Statement> ConfiguredStatement<T> inject(
      final ConfiguredStatement<T> statement
  ) {
    return inject(statement, new TopicProperties.Builder());
  }
@SuppressWarnings("unchecked")
  @Test
  public void shouldBuildWithClauseWithTopicProperties() {
    // Given:
    givenStatement("CREATE STREAM x WITH (kafka_topic='topic') AS SELECT * FROM SOURCE;");

    when(builder.build()).thenReturn(new TopicProperties("expectedName", 10, (short) 10, (long) 5000));

    // When:
    final ConfiguredStatement<CreateAsSelect> result =
        (ConfiguredStatement<CreateAsSelect>) injector.inject(statement, builder);

    // Then:
    final CreateSourceAsProperties props = result.getStatement().getProperties();
    assertThat(props.getKafkaTopic(), is(Optional.of("expectedName")));
    assertThat(props.getPartitions(), is(Optional.of(10)));
    assertThat(props.getReplicas(), is(Optional.of((short) 10)));
    assertThat(props.getRetentionInMillis(), is(Optional.of((long) 5000)));
  }