[#24971] Add a retry policy for JmsIO #24971

[Amraneze]

Add a retry policy to publish messages in case of connection closed

Fixes #24971

Co-Authored-By: Amrane Ait Zeouay [email protected]

When trying to publish messages using JmsIO with a closed/failed connection, the messages are not published and lost. We will need to get the failed messages with the function getFailedMessages and try again and again multiple times.

var failedPublishedMessages = messages.apply(getWriter(sinkOptions)).getFailedMessages();
var refailedPublishedMessages = failedPublishedMessages.apply(REPUBLISH_FAILED_MESSAGE_NAME, getWriter(sinkOptions)).getFailedMessages();
....
return PDone.in(cdpMessages.getPipeline());

To not have multiple steps for retrying the failed publication, we added a Retry Policy which consist on retrying the publication n times (the number is given by the user) until the timeout (given by the user too) is finished. Also, we added a predicate function, to give the user the option to handle when the connection should be reconnected or not in case of a failed connection.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[github-actions]

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).

[Abacn]

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.

[Amraneze]

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.

[Abacn]

Thanks for the change! I see what has caused the issue, and commented above that there are changes still needed in order to fix it. Please let me know if it is not clear to you or if there is any questions.

[github-actions]

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

[Abacn]

Thanks for the change. got a couple of comments should be easy to resolve. Thanks for your patience.

[Amraneze]

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 ?

[github-actions]

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)

[apilloud]

R: @Abacn

[Abacn]

Sorry for the delay. Will take another look tomorrow.

[github-actions]

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

[Abacn]

Thanks, it's getting close. Found a bug and a design suggestion here.

[Abacn]

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

[Amraneze]

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.

[Amraneze]

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