public static <EventT> Write<EventT> write() {
    return new AutoValue_JmsIO_Write.Builder<EventT>().build();
  }
@Test
  public void testWriteMessageWithRetryPolicyReachesLimit() throws Exception {
    String messageText = "text";
    int maxPublicationAttempts = 2;
    List<String> data = Collections.singletonList(messageText);
    RetryConfiguration retryConfiguration =
        RetryConfiguration.create(maxPublicationAttempts, null, null);

    WriteJmsResult<String> output =
        pipeline
            .apply(Create.of(data))
            .apply(
                JmsIO.<String>write()
                    .withConnectionFactory(connectionFactory)
                    .withValueMapper(
                        (SerializableBiFunction<String, Session, Message>)
                            (s, session) -> {
                              throw new JmsIOException("Error!!");
                            })
                    .withRetryConfiguration(retryConfiguration)
                    .withQueue(QUEUE)
                    .withUsername(USERNAME)
                    .withPassword(PASSWORD));

    PAssert.that(output.getFailedMessages()).containsInAnyOrder(messageText);
    PipelineResult pipelineResult = pipeline.run();

    MetricQueryResults metrics =
        pipelineResult
            .metrics()
            .queryMetrics(
                MetricsFilter.builder()
                    .addNameFilter(
                        MetricNameFilter.named(
                            JMS_IO_PRODUCER_METRIC_NAME, PUBLICATION_RETRIES_METRIC_NAME))
                    .build());

    assertThat(
        metrics.getCounters(),
        contains(
            allOf(
                hasProperty("attempted", is((long) maxPublicationAttempts)),
                hasProperty(
                    "key",
                    hasToString(
                        containsString(
                            String.format(
                                "%s:%s",
                                JMS_IO_PRODUCER_METRIC_NAME, PUBLICATION_RETRIES_METRIC_NAME)))))));

    assertQueueIsEmpty();
  }