### Question:
SqsStreamProvisioner implements
                                  ProvisioningProvider<ExtendedConsumerProperties<SqsConsumerProperties>, ExtendedProducerProperties<SqsProducerProperties>> { @Override public ProducerDestination provisionProducerDestination(String name, ExtendedProducerProperties<SqsProducerProperties> properties) throws ProvisioningException { CreateTopicResult createTopicResult = amazonSNSAsync.createTopic(name); return new SqsProducerDestination(name, createTopicResult.getTopicArn()); } SqsStreamProvisioner(AmazonSQSAsync amazonSQSAsync, AmazonSNSAsync amazonSNSAsync); @Override ProducerDestination provisionProducerDestination(String name,
                                                            ExtendedProducerProperties<SqsProducerProperties> properties); @Override ConsumerDestination provisionConsumerDestination(String name, String group,
                                                            ExtendedConsumerProperties<SqsConsumerProperties> properties);  }### Answer:
@Test public void createsSnsTopicWhenProvisionsProducer() { ExtendedProducerProperties<SqsProducerProperties> producerProperties = new ExtendedProducerProperties<>(new SqsProducerProperties()); sqsStreamProvisioner.provisionProducerDestination("topic-name", producerProperties); List<String> topicArns = listTopicsArns(); assertThat(topicArns).hasSize(1); assertThat(sns.getTopicAttributes(topicArns.get(0)).getAttributes().get("TopicArn")).endsWith("topic-name"); }