Abacn commented on code in PR #24973:
URL: https://github.com/apache/beam/pull/24973#discussion_r1103119672
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java:
##########
@@ -932,36 +1036,158 @@ 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);
+ this.isProducerNeedsToBeCreated = false;
}
}
- @ProcessElement
- public void processElement(ProcessContext ctx) {
+ public void publishMessage(T input) throws JMSException, JmsIOException {
Destination destinationToSendTo = destination;
try {
- Message message = spec.getValueMapper().apply(ctx.element(),
session);
+ Message message = spec.getValueMapper().apply(input, session);
if (spec.getTopicNameMapper() != null) {
- destinationToSendTo =
-
session.createTopic(spec.getTopicNameMapper().apply(ctx.element()));
+ destinationToSendTo =
session.createTopic(spec.getTopicNameMapper().apply(input));
}
producer.send(destinationToSendTo, message);
- } catch (Exception ex) {
- LOG.error("Error sending message on topic {}", destinationToSendTo);
- ctx.output(failedMessageTag, ctx.element());
+ } catch (JMSException | JmsIOException | NullPointerException
exception) {
+ // Handle NPE in case of getValueMapper or getTopicNameMapper
returns NPE
+ if (exception instanceof NullPointerException) {
+ throw new JmsIOException("An error occurred", exception);
+ }
+ throw exception;
+ }
+ }
+
+ public void close() throws JMSException {
+ isProducerNeedsToBeCreated = true;
+ if (producer != null) {
+ producer.close();
+ producer = null;
+ }
+ if (session != null) {
+ session.close();
+ session = null;
}
+ if (connection != null) {
+ try {
+ // If the connection failed, stopping the connection will throw a
JMSException
+ connection.stop();
+ } catch (JMSException exception) {
+ LOG.warn("The connection couldn't be closed", exception);
+ }
+ connection.close();
+ connection = null;
+ }
+ }
+ }
+
+ private static class JmsIOProducerFn<T> extends DoFn<T, T> {
+
+ private final @Initialized JmsConnection<T> jmsConnection;
+ private final TupleTag<T> failedMessagesTag;
+
+ JmsIOProducerFn(JmsIO.Write<T> spec, TupleTag<T> failedMessagesTag) {
+ this.failedMessagesTag = failedMessagesTag;
+ this.jmsConnection = new JmsConnection<>(spec);
+ }
+
+ @StartBundle
+ public void startBundle() throws JMSException {
+ this.jmsConnection.start();
+ }
+
+ @ProcessElement
+ public void processElement(@Element T input, ProcessContext context) {
+ try {
+ this.jmsConnection.publishMessage(input);
+ } catch (JMSException | JmsIOException exception) {
+ context.output(this.failedMessagesTag, input);
+ }
+ }
+
+ @FinishBundle
+ public void finishBundle() throws JMSException {
+ this.jmsConnection.close();
+ }
+
+ @Teardown
+ public void tearDown() throws JMSException {
+ this.jmsConnection.close();
+ }
+ }
+
+ static class JmsIOProduceRetryFn<T> extends DoFn<T, T> {
+
+ private transient @Initialized FluentBackoff retryBackOff;
+
+ private final JmsIO.Write<T> spec;
+ private final TupleTag<T> failedMessagesTags;
+ private final @Initialized JmsConnection<T> jmsConnection;
+ private final Counter publicationRetries =
+ Metrics.counter(JMS_IO_PRODUCER_METRIC_NAME,
PUBLICATION_RETRIES_METRIC_NAME);
+
+ JmsIOProduceRetryFn(JmsIO.Write<T> spec, TupleTag<T> failedMessagesTags)
{
+ this.spec = spec;
+ this.failedMessagesTags = failedMessagesTags;
+ this.jmsConnection = new JmsConnection<>(spec);
+ }
+
+ @Setup
+ public void setup() {
+ RetryConfiguration retryConfiguration =
checkStateNotNull(spec.getRetryConfiguration());
+ retryBackOff =
+ FluentBackoff.DEFAULT
+
.withInitialBackoff(checkStateNotNull(retryConfiguration.getInitialDuration()))
+
.withMaxCumulativeBackoff(checkStateNotNull(retryConfiguration.getMaxDuration()))
+ .withMaxRetries(retryConfiguration.getMaxAttempts());
+ }
+
+ @StartBundle
+ public void startBundle() throws JMSException {
+ this.jmsConnection.start();
+ }
+
+ @ProcessElement
+ public void processElement(@Element T input, ProcessContext context) {
+ try {
+ publishMessage(input, context);
+ } catch (IOException | InterruptedException exception) {
+ LOG.error("Error while publishing the message", exception);
Review Comment:
yes that looks good
--
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]
