SnsIO { public static <T> Write<T> write() { return new AutoValue_SnsIO_Write.Builder().build(); }  static Write<T> write(); static WriteAsync<T> writeAsync();  }
@Test public void testDataWritesToSNS() { ImmutableList<String> input = ImmutableList.of("message1", "message2"); final PCollection<PublishResponse> results = p.apply(Create.of(input)) .apply( SnsIO.<String>write() .withPublishRequestFn(SnsIOTest::createSampleMessage) .withTopicArn(topicArn) .withRetryConfiguration( SnsIO.RetryConfiguration.create( 5, org.joda.time.Duration.standardMinutes(1))) .withSnsClientProvider(SnsClientMockSuccess::new)); final PCollection<Long> publishedResultsSize = results.apply(Count.globally()); PAssert.that(publishedResultsSize).containsInAnyOrder(ImmutableList.of(2L)); p.run().waitUntilFinish(); }
@Test public void testRetries() throws Throwable { thrown.expectMessage("Error writing to SNS"); ImmutableList<String> input = ImmutableList.of("message1", "message2"); p.apply(Create.of(input)) .apply( SnsIO.<String>write() .withPublishRequestFn(SnsIOTest::createSampleMessage) .withTopicArn(topicArn) .withRetryConfiguration( SnsIO.RetryConfiguration.create(4, org.joda.time.Duration.standardSeconds(10))) .withSnsClientProvider(SnsClientMockErrors::new)); try { p.run(); } catch (final Pipeline.PipelineExecutionException e) { expectedLogs.verifyWarn(String.format(SnsIO.Write.SnsWriterFn.RETRY_ATTEMPT_LOG, 1)); expectedLogs.verifyWarn(String.format(SnsIO.Write.SnsWriterFn.RETRY_ATTEMPT_LOG, 2)); expectedLogs.verifyWarn(String.format(SnsIO.Write.SnsWriterFn.RETRY_ATTEMPT_LOG, 3)); throw e.getCause(); } fail("Pipeline is expected to fail because we were unable to write to SNS."); }