[Bug]: Messages are not published when a connection is closed with JmsIO

[Amraneze]

What happened?

When using JmsIO for publishing messages, we found out that messages are not being published when the connection is closed due to any reason (Network, Broker down ...). Even though we have the option to get failed published messages we don't have a way to have a retry policy.

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

Note: the function getWriter returns JMS writer JmsIO.Write<Message>

Issue Priority

Priority: 1 (data loss / total loss of function)

Issue Components

• Component: IO connector

[kennknowles]

Does it cause a failure of the processing of the element? In that case the runner should retry the processing with a fresh instance of the DoFn, which should retry the publish. It isn't as good as a retry policy, but it avoids data loss.

[Amraneze]

Correct me if I'm wrong, but in a bundle, when a message is failed the whole bundle will be retried which is not what we want too also, the exception was catched in processElement which makes the DoFn to keep the same instance of the publisher with a failed connection.