Amraneze commented on code in PR #24973:
URL: https://github.com/apache/beam/pull/24973#discussion_r1087195879
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java:
##########
@@ -932,36 +1067,78 @@ public void setup() throws Exception {
} else if (spec.getTopic() != null) {
this.destination = session.createTopic(spec.getTopic());
}
-
- this.producer = this.session.createProducer(null);
+ this.producer = this.session.createProducer(this.destination);
}
}
@ProcessElement
- public void processElement(ProcessContext ctx) {
- Destination destinationToSendTo = destination;
+ public void processElement(@Element T input, ProcessContext context) {
try {
- Message message = spec.getValueMapper().apply(ctx.element(),
session);
- if (spec.getTopicNameMapper() != null) {
- destinationToSendTo =
-
session.createTopic(spec.getTopicNameMapper().apply(ctx.element()));
+ publishMessage(input, context);
+ } catch (IOException | InterruptedException exception) {
+ LOG.error("Error while publishing the message", exception);
+ context.output(failedMessagesTag, input);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void publishMessage(T input, ProcessContext context)
+ throws IOException, InterruptedException {
+ Sleeper sleeper = Sleeper.DEFAULT;
+ Destination destinationToSendTo = destination;
+ BackOff backoff = checkStateNotNull(retryBackOff).backoff();
+ int publicationAttempt = 0;
+ while (publicationAttempt >= 0) {
+ publicationAttempt++;
+ try {
+ Message message = spec.getValueMapper().apply(input, session);
+ if (spec.getTopicNameMapper() != null) {
+ destinationToSendTo =
session.createTopic(spec.getTopicNameMapper().apply(input));
+ }
+ producer.send(destinationToSendTo, message);
+ publicationAttempt = -1;
+ } catch (Exception exception) {
+ if (!BackOffUtils.next(sleeper, backoff)) {
+ LOG.error("The message wasn't published to topic {}",
destinationToSendTo, exception);
+ context.output(failedMessagesTag, input);
+ publicationAttempt = -1;
+ } else {
+ publicationRetries.inc();
+ LOG.warn(
+ "Error sending message on topic {}, retry attempt {}",
+ destinationToSendTo,
+ publicationAttempt,
+ exception);
+ }
}
- producer.send(destinationToSendTo, message);
- } catch (Exception ex) {
- LOG.error("Error sending message on topic {}", destinationToSendTo);
- ctx.output(failedMessageTag, ctx.element());
}
}
+ private void restartJmsConnection() throws JMSException {
Review Comment:
When the connection is lost or closed, the JMS client calls the
`ExceptionListener` if it exists. I added that listener
````java
this.connection.setExceptionListener(
exception -> {
if (spec.getConnectionFailedPredicate() != null &&
spec.getConnectionFailedPredicate().apply(exception)) {
LOG.error("Jms connection encountered the following exception:",
exception);
connectionErrors.inc();
try {
restartJmsConnection();
} catch (JMSException e) {
LOG.error("An error occurred while reconnecting:", e);
}
}
});
````
Should I just re-throw the exception so that the DoFn will call the
`FinishBundle` method ? Something like this
````java
this.connection.setExceptionListener(exception -> {
LOG.error("Jms connection encountered the following exception:",
exception);
connectionErrors.inc();
throw new JmsIOException("The connection was closed", exception);
});
````
Or what do you think we should do?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
