Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @robertwb for label java.
R: @johnjcasey for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

just fyi beam has backoff utility classes for retry:
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BackOffUtils.java
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java

EDIT: never mind, I see the implenetation is using FluentBackoff.
Is there a need to create a separate class for RetryPolicy? It is only used in JmsProducer and most likely users do not to set them. Generally api calls should have a retry mechanism by default.

just fyi beam has backoff utility classes for retry: https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BackOffUtils.java https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java

EDIT: never mind, I see the implenetation is using FluentBackoff. Is there a need to create a separate class for RetryPolicy? It is only used in JmsProducer and most likely users do not to set them. Generally api calls should have a retry mechanism by default.

The issue is that JmsIO doesn't retry at all when publishing a message, any message that is failed to publish is pushed to ouput as you can see in JmsIO code:

@ProcessElement
public void processElement(ProcessContext ctx) {
  Destination destinationToSendTo = destination;
  try {
    Message message = spec.getValueMapper().apply(ctx.element(), session);
    if (spec.getTopicNameMapper() != null) {
      destinationToSendTo =
          session.createTopic(spec.getTopicNameMapper().apply(ctx.element()));
    }
    producer.send(destinationToSendTo, message);
  } catch (Exception ex) {
    LOG.error("Error sending message on topic {}", destinationToSendTo);
    ctx.output(failedMessageTag, ctx.element());
  }
}

If there are any exceptions, they will be catched and the message is pushed to failed message tag's output. Which means we will need to get the failed messages and retry by ourselves N times like this:

public PDone expand(PCollection<Message> messages) {
    // Retry 3 times when the session is closed
    messages.apply(getJmsWriter(sinkOptions))
          .getFailedMessages()
         .apply(REPUBLISH_FAILED_MESSAGE_NAME, getJmsWriter(sinkOptions))
           .getFailedMessages()
         .apply(REPUBLISH_FAILED_MESSAGE_NAME, getJmsWriter(sinkOptions))
           .getFailedMessages()
         .apply(REPUBLISH_FAILED_MESSAGE_NAME, getJmsWriter(sinkOptions));

    return PDone.in(messages.getPipeline());
  }

  private static JmsIO.Write<Message> getJmsWriter(SinkOptions sinkOptions) {
    return JmsIO.<Message>write()
        .withConnectionFactory(SinkOptions.getConnectionFactory())
        .withValueMapper(getValueMapper())
        .withTopicNameMapper(getTopicNameMapper());
  }

When the session is closed and the first PTransform that is publishing the message is displaying error logs Error sending message on topic, it will not reconnect and just send all messages to the second step as you can see in the screenshot

Note: The graph of first PTransform is the same as the second PTransform after the session is closed, which means the first step doesn't publish at all and it just send all messages to the second step.

Reminder, please take a look at this pr: @robertwb @johnjcasey

Hello @Abacn, Sorry I was busy with work. I redesigned the flow of JmsIOWriter. I tried to create a cache and flush all the messages in finishBundle but we will need to handle the exception in processElement otherwise the DoFn will tear down without calling finishBundle. I added a new step which will be used only for processing failed published messages and also I added a condition whenever the producer needs to be recreated it. Let me know what do you think and if is there any tests to add ?

PS: Should I add a benchmark test ?

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @apilloud for label java.
R: @Abacn for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

R: @Abacn

Sorry for the delay. Will take another look tomorrow.

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

Thanks. This LGTM.

nit: the added back-off unit tests take a while to complete. In the case of test may use a mocked sleeper

I will improve the test with the new API calls.

@Abacn can you please merge the PR, I don't have access to merge it :( ?