@Override
  public <T extends Statement> ConfiguredStatement<T> inject(
      final ConfiguredStatement<T> statement
  ) {
    return inject(statement, new TopicProperties.Builder());
  }
@Test
  public void shouldPassThroughWithClauseToBuilderForCreate() {
    // Given:
    givenStatement("CREATE STREAM x (FOO VARCHAR) WITH(value_format='avro', kafka_topic='topic', partitions=2, retention_ms=5000);");

    final CreateSourceProperties props = ((CreateSource) statement.getStatement())
        .getProperties();

    // When:
    injector.inject(statement, builder);

    // Then:

    verify(builder).withWithClause(
        Optional.of(props.getKafkaTopic()),
        props.getPartitions(),
        props.getReplicas(),
        props.getRetentionInMillis()
    );

  }