public static <T> Write<T> write() {
    return new AutoValue_SnsIO_Write.Builder<T>()
        .setClientConfiguration(ClientConfiguration.builder().build())
        .build();
  }
@Test
  public void testWriteWithoutTopicArn() {
    List<String> input = ImmutableList.of("message1", "message2");

    when(sns.publish(any(PublishRequest.class)))
        .thenReturn(PublishResponse.builder().messageId("id").build());

    Write<String> snsWrite =
        SnsIO.<String>write().withPublishRequestBuilder(msg -> requestBuilder(msg, topicArn));

    PCollection<PublishResponse> results = p.apply(Create.of(input)).apply(snsWrite);
    PAssert.that(results.apply(Count.globally())).containsInAnyOrder(2L);
    p.run();

    verify(sns, times(0)).getTopicAttributes(any(Consumer.class));
    for (String msg : input) {
      verify(sns).publish(requestBuilder(msg, topicArn).build());
    }
  }