### Question:
KinesisSourceProvider implements SourceProvider { @Override public <T> SourceFunction<T> forSpec(IngressSpec<T> spec) { final KinesisIngressSpec<T> kinesisIngressSpec = asKinesisSpec(spec); return new FlinkKinesisConsumer<>( kinesisIngressSpec.streams(), deserializationSchemaFromSpec(kinesisIngressSpec), propertiesFromSpec(kinesisIngressSpec)); }  @Override SourceFunction<T> forSpec(IngressSpec<T> spec);  }### Answer:
@Test public void exampleUsage() { final KinesisIngressSpec<String> kinesisIngressSpec = KinesisIngressBuilder.forIdentifier(ID) .withAwsRegion("us-west-1") .withAwsCredentials(AwsCredentials.basic("access-key-id", "secret-access-key")) .withDeserializer(TestDeserializer.class) .withStream(STREAM_NAME) .build(); final KinesisSourceProvider provider = new KinesisSourceProvider(); final SourceFunction<String> source = provider.forSpec(kinesisIngressSpec); assertThat(source, instanceOf(FlinkKinesisConsumer.class)); }