### Question:
KinesisSinkProvider implements SinkProvider { @Override public <T> SinkFunction<T> forSpec(EgressSpec<T> spec) { final KinesisEgressSpec<T> kinesisEgressSpec = asKinesisSpec(spec); final CachingPartitionerSerializerDelegate<T> partitionerSerializerDelegate = new CachingPartitionerSerializerDelegate<>(serializerInstanceFromSpec(kinesisEgressSpec)); final FlinkKinesisProducer<T> kinesisProducer = new FlinkKinesisProducer<>( partitionerSerializerDelegate, propertiesFromSpec(kinesisEgressSpec)); kinesisProducer.setCustomPartitioner(partitionerSerializerDelegate); kinesisProducer.setQueueLimit(kinesisEgressSpec.maxOutstandingRecords()); kinesisProducer.setFailOnError(true); return kinesisProducer; }  @Override SinkFunction<T> forSpec(EgressSpec<T> spec);  }### Answer:
@Test public void exampleUsage() { final KinesisEgressSpec<String> kinesisEgressSpec = KinesisEgressBuilder.forIdentifier(ID) .withAwsRegion("us-west-1") .withAwsCredentials(AwsCredentials.basic("access-key-id", "secret-access-key")) .withSerializer(TestSerializer.class) .build(); final KinesisSinkProvider provider = new KinesisSinkProvider(); final SinkFunction<String> sink = provider.forSpec(kinesisEgressSpec); assertThat(sink, instanceOf(FlinkKinesisProducer.class)); }