davsclaus commented on code in PR #27173: URL: https://github.com/apache/camel/pull/27173#discussion_r4236603668
########## components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerTransactedTest.java: ########## @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.sjms.batch; + +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.IntStream; + +import org.apache.camel.Processor; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.sjms.support.JmsTestSupport; +import org.junit.jupiter.api.Test; + +import static java.lang.String.format; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.BATCH_ROUTEBUILDER_MOCK_FINISH; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.BATCH_ROUTEBUILDER_MOCK_START; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.assertBatchSizesInOrder; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.createBatchRoute; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.getBatchBodiesAsString; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.sendMessagesWithText; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class BatchConsumerTransactedTest extends JmsTestSupport { + + private static final String QUEUE_NAME_TEMPLATE = "batch.consumer.%s.BatchConsumerTransactedTest"; + + private static final String ROUTE_ID_SESSION_TX = "tx"; + private static final String ROUTE_ID_CLIENT_ACK_NO_TX = "no-tx-client-ack"; + private static final String ROUTE_ID_AUTO_ACK_NO_TX = "no-tx-auto"; + + private static final String MESSAGE_TEXT = "Message %d"; + + @Test + public void testSessionTransacted() throws Exception { + MockEndpoint mockStart = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_START, ROUTE_ID_SESSION_TX)); + mockStart.expectedMessageCount(2); + + MockEndpoint mockFinish = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_FINISH, ROUTE_ID_SESSION_TX)); + mockFinish.expectedMessageCount(1); + + sendMessagesWithText(template, format("sjms:queue:" + QUEUE_NAME_TEMPLATE, ROUTE_ID_SESSION_TX), + 5, MESSAGE_TEXT); + + MockEndpoint.assertIsSatisfied(context); + assertBatchSizesInOrder(mockFinish, 5); + assertEquals( + IntStream.rangeClosed(1, 5) + .mapToObj(i -> String.format(MESSAGE_TEXT, i)) + .toList(), + getBatchBodiesAsString(mockFinish.getExchanges().get(0))); + } + + @Test + public void testClientAcknowledgedNotTransacted() throws Exception { + MockEndpoint mockStart = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_START, ROUTE_ID_CLIENT_ACK_NO_TX)); + mockStart.expectedMessageCount(2); + + MockEndpoint mockFinish = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_FINISH, ROUTE_ID_CLIENT_ACK_NO_TX)); + mockFinish.expectedMessageCount(1); + + sendMessagesWithText(template, + format("sjms:queue:" + QUEUE_NAME_TEMPLATE, ROUTE_ID_CLIENT_ACK_NO_TX), + 5, MESSAGE_TEXT); + + MockEndpoint.assertIsSatisfied(context); + assertBatchSizesInOrder(mockFinish, 5); + assertEquals( + IntStream.rangeClosed(1, 5) + .mapToObj(i -> String.format(MESSAGE_TEXT, i)) + .toList(), + getBatchBodiesAsString(mockFinish.getExchanges().get(0))); + } + + @Test + public void testAutoAcknowledgedNotTransacted() throws Exception { + MockEndpoint mockStart = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_START, ROUTE_ID_AUTO_ACK_NO_TX)); + mockStart.expectedMessageCount(1); + + MockEndpoint mockFinish = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_FINISH, ROUTE_ID_AUTO_ACK_NO_TX)); + mockFinish.expectedMessageCount(0); + + sendMessagesWithText(template, + format("sjms:queue:" + QUEUE_NAME_TEMPLATE, ROUTE_ID_AUTO_ACK_NO_TX), 5, MESSAGE_TEXT); + + MockEndpoint.assertIsSatisfied(context); + } + + @Override + protected RoutesBuilder[] createRouteBuilders() { + return new RoutesBuilder[] { + createBatchRoute(QUEUE_NAME_TEMPLATE, ROUTE_ID_SESSION_TX, 5, + 1000, + true, null, 1, new ThrowExceptionProcessor()), + createBatchRoute(QUEUE_NAME_TEMPLATE, ROUTE_ID_CLIENT_ACK_NO_TX, + 5, 1000, false, "CLIENT_ACKNOWLEDGE", 1, new ThrowExceptionProcessor()), + createBatchRoute(QUEUE_NAME_TEMPLATE, ROUTE_ID_AUTO_ACK_NO_TX, 5, 1000, false, + "AUTO_ACKNOWLEDGE", 1, new ThrowExceptionProcessor()) + }; + } + + private static class ThrowExceptionProcessor implements Processor { + private final AtomicInteger counter = new AtomicInteger(); + + @Override + public void process(org.apache.camel.Exchange exchange) { Review Comment: `org.apache.camel.Exchange` is still fully qualified here, and on line 87 of `BatchConsumerRollbackOnlyTest`. Neither file imports `Exchange` yet, so please add `import org.apache.camel.Exchange;` and use `Exchange exchange`. The build's OpenRewrite step shortens these FQNs anyway, which would show up as uncommitted changes in CI. ########## components/camel-sjms/src/main/docs/sjms-component.adoc: ########## @@ -312,3 +312,32 @@ Currently, the only correlation strategy is to use the `JMSCorrelationId`. The _InOut_ Consumer uses this strategy as well ensuring that all response messages to the included `JMSReplyTo` destination also have the `JMSCorrelationId` copied from the request as well. + +=== Batch consuming + +The consumer can be configured to receive messages in batches instead of one at a time, by setting +`batching=true`. Instead of one Exchange per message, the route receives a single Exchange whose body +is a `List<Exchange>`, one per JMS message in the batch, with a `CamelSjmsBatchSize` header giving the +batch's size. Each Exchange in the batch is comparable to the Exchange generated by a non-batching consumer. + +Batching only supports the InOnly exchange pattern. There is no reply-to/request-response support for +batched consumption — a batch of N unrelated messages has no well-defined single reply, so InOut +routes are not supported when `batching=true`. + +A batch completes when either `batchSize` messages have been received, or the `batchInterval` — +measured from the first message in the batch - has elapsed. + +Commit/acknowledgement happens once per batch rather than once per message. With +`acknowledgementMode=CLIENT_ACKNOWLEDGE` if the whole batch should be redelivered together on failure. Review Comment: When the suggestion was applied, the two lines about `AUTO_ACKNOWLEDGE` were dropped, so the paragraph now reads "With `acknowledgementMode=CLIENT_ACKNOWLEDGE` if the whole batch should be redelivered...". Restoring them: ```suggestion `acknowledgementMode=AUTO_ACKNOWLEDGE`, messages are acknowledged as they're received into the batch, before routing, so a failed batch cannot be redelivered. Use `transacted=true` or `acknowledgementMode=CLIENT_ACKNOWLEDGE` if the whole batch should be redelivered together on failure. ``` The file also no longer ends with a newline (after line 343). ########## components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/BatchConsumerWorker.java: ########## @@ -0,0 +1,181 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.sjms.consumer; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import jakarta.jms.JMSException; +import jakarta.jms.Message; +import jakarta.jms.MessageConsumer; +import jakarta.jms.Session; + +import org.apache.camel.component.sjms.SjmsEndpoint; +import org.apache.camel.component.sjms.jms.SessionAcknowledgementType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.apache.camel.component.sjms.SjmsHelper.commitIfNeeded; +import static org.apache.camel.component.sjms.SjmsHelper.rollbackIfNeeded; + +public class BatchConsumerWorker implements Runnable { + + private static final Logger LOG = LoggerFactory.getLogger(BatchConsumerWorker.class); + + private static final long JMS_CONSUMER_RECEIVE_WAKE_INTERVAL_TIMEOUT = 1000L; + private static final long JMS_CONSUMER_RECEIVE_MIN_TIMEOUT = 100L; + + private final SjmsEndpoint endpoint; + private final BatchEndpointMessageListener batchListener; + private final MessageConsumer consumer; + private final Session session; + private final AtomicBoolean running = new AtomicBoolean(true); + private volatile boolean jmsUnhealthy; + + BatchConsumerWorker(SjmsEndpoint endpoint, BatchEndpointMessageListener batchListener, + Session session, MessageConsumer consumer) { + this.endpoint = endpoint; + this.batchListener = batchListener; + this.session = session; + this.consumer = consumer; + } + + void shutdown() { + LOG.debug("Shutdown requested"); + running.set(false); + } + + boolean isShutdownRequested() { + return !running.get(); + } + + void invalidate() { + jmsUnhealthy = true; + shutdown(); + } + + private boolean isRedeliverable() { + return endpoint.isTransacted() + || endpoint.getAcknowledgementMode() == SessionAcknowledgementType.CLIENT_ACKNOWLEDGE; + } + + @Override + public void run() { + LOG.debug("run START"); + int batchSize = endpoint.getBatchSize(); + long batchInterval = endpoint.getBatchInterval(); + List<Message> buffer = new ArrayList<>(); + long batchStartTime = 0L; + + try { + while (running.get()) { + long waitMillis + = computeWaitMillis(buffer.isEmpty(), batchStartTime, batchInterval); + + Message msg = consumer.receive(waitMillis); + + if (msg != null) { + if (buffer.isEmpty()) { + batchStartTime = System.nanoTime(); + } + buffer.add(msg); + } + + boolean sizeReached = batchSize > 0 && buffer.size() >= batchSize; + boolean intervalElapsed = batchInterval > 0 && !buffer.isEmpty() + && TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - batchStartTime) >= batchInterval; + + if (sizeReached || intervalElapsed) { + dispatchOrDiscard(buffer); + buffer = new ArrayList<>(); + } + } + + if (!buffer.isEmpty()) { + dispatchOrDiscard(buffer); // graceful-stop drain + } + } catch (JMSException e) { + jmsUnhealthy = true; + if (!buffer.isEmpty()) { + dispatchOrDiscard(buffer); + } + throw new BatchConsumerWorkerException(e); + } + LOG.info("run STOP"); + } + + private void dispatchOrDiscard(List<Message> batch) { + if (jmsUnhealthy) { + if (isRedeliverable()) { + LOG.warn("JMS is unhealthy on {}: discarding {} buffered message(s); " + + "unacknowledged/uncommitted, will be redelivered", + endpoint.getEndpointUri(), batch.size()); + return; + } + LOG.warn("JMS is unhealthy on {}: attempting best-effort dispatch of {} already-acknowledged " + + "message(s) since they cannot be redelivered", + endpoint.getEndpointUri(), batch.size()); + } + onBatch(batch); + } + + private void onBatch(List<Message> batch) { + try { + doOnBatch(batch); + } catch (Exception e) { + if (e instanceof JMSException jmsException) { + if (endpoint.getExceptionListener() != null) { + endpoint.getExceptionListener().onException(jmsException); + } + } else { + batchListener.handleException( + "Execution of JMS message listener failed. This exception is ignored.", + e); + LOG.warn("Execution of JMS message listener failed. This exception is ignored.", e); Review Comment: Thanks for wiring this in. As gnodet-bot noted, the default `LoggingExceptionHandler` already logs at WARN, so the `LOG.warn` logs every failure a second time. "This exception is ignored" also isn't true anymore: the batch has been rolled back and will be redelivered. For example: ```suggestion batchListener.handleException("Error processing batch, the batch is rolled back", e); ``` ########## components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerTransactedTest.java: ########## @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.sjms.batch; + +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.IntStream; + +import org.apache.camel.Processor; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.sjms.support.JmsTestSupport; +import org.junit.jupiter.api.Test; + +import static java.lang.String.format; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.BATCH_ROUTEBUILDER_MOCK_FINISH; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.BATCH_ROUTEBUILDER_MOCK_START; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.assertBatchSizesInOrder; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.createBatchRoute; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.getBatchBodiesAsString; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.sendMessagesWithText; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class BatchConsumerTransactedTest extends JmsTestSupport { + + private static final String QUEUE_NAME_TEMPLATE = "batch.consumer.%s.BatchConsumerTransactedTest"; + + private static final String ROUTE_ID_SESSION_TX = "tx"; + private static final String ROUTE_ID_CLIENT_ACK_NO_TX = "no-tx-client-ack"; + private static final String ROUTE_ID_AUTO_ACK_NO_TX = "no-tx-auto"; + + private static final String MESSAGE_TEXT = "Message %d"; + + @Test + public void testSessionTransacted() throws Exception { + MockEndpoint mockStart = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_START, ROUTE_ID_SESSION_TX)); + mockStart.expectedMessageCount(2); + + MockEndpoint mockFinish = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_FINISH, ROUTE_ID_SESSION_TX)); + mockFinish.expectedMessageCount(1); + + sendMessagesWithText(template, format("sjms:queue:" + QUEUE_NAME_TEMPLATE, ROUTE_ID_SESSION_TX), + 5, MESSAGE_TEXT); + + MockEndpoint.assertIsSatisfied(context); + assertBatchSizesInOrder(mockFinish, 5); + assertEquals( + IntStream.rangeClosed(1, 5) + .mapToObj(i -> String.format(MESSAGE_TEXT, i)) + .toList(), + getBatchBodiesAsString(mockFinish.getExchanges().get(0))); Review Comment: The distinct bodies help. To show that the redelivered batch holds the same messages as the failed one, compare the two attempts, and use the so-far unused `assertBatchRedelivered`. For example: ```java assertEquals(getBatchBodiesAsString(mockStart.getExchanges().get(0)), getBatchBodiesAsString(mockFinish.getExchanges().get(0))); assertBatchRedelivered(mockFinish.getExchanges().get(0), "Message 1", true); ``` (The same applies to `testClientAcknowledgedNotTransacted`.) Otherwise `assertBatchRedelivered` in `BatchTestHelper` can be removed. There is still no test for the new `ExceptionHandler` path. A route with `transacted=true` and `exceptionHandler=#myHandler` (a handler that records its calls), whose processor always fails, would check that the handler is called and that the batch is redelivered. ########## components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/BatchEndpointMessageListener.java: ########## @@ -0,0 +1,111 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.sjms.consumer; + +import java.util.ArrayList; +import java.util.List; + +import jakarta.jms.Message; +import jakarta.jms.Session; + +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.RollbackExchangeException; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.component.sjms.SjmsConstants; +import org.apache.camel.component.sjms.SjmsConsumer; +import org.apache.camel.component.sjms.SjmsEndpoint; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.apache.camel.RuntimeCamelException.wrapRuntimeCamelException; + +public class BatchEndpointMessageListener { + + private static final Logger LOG = LoggerFactory.getLogger(BatchEndpointMessageListener.class); + + private final SjmsConsumer consumer; + private final SjmsEndpoint endpoint; + private final Processor processor; + + public BatchEndpointMessageListener(SjmsConsumer consumer, SjmsEndpoint endpoint, Processor processor) { + this.consumer = consumer; + this.endpoint = endpoint; + this.processor = processor; + } + + private Exchange aggregate(List<Message> rawMessages, Session session) { + List<Exchange> exchanges = new ArrayList<>(rawMessages.size()); + for (Message m : rawMessages) { + Exchange e = endpoint.createExchange(m, session); + // Force eager materialization of JMS message headers and body into the + // Camel Exchange, before the session is committed/closed after dispatch. + e.getIn().getHeaders(); + e.getIn().getBody(); + + exchanges.add(e); + } + + Exchange batchExchange = consumer.createExchange(false); + batchExchange.getIn().setBody(exchanges); + batchExchange.setProperty(SjmsConstants.JMS_SESSION, session); + batchExchange.getIn().setHeader(SjmsConstants.SJMS_BATCH_SIZE_HEADER, rawMessages.size()); + + return batchExchange; + } + + void onBatch(List<Message> rawMessages, Session session) throws Exception { + LOG.trace("onBatch START"); + + LOG.debug("{} consumer received batch message: {}", endpoint, rawMessages); + Exchange batchExchange = null; + RuntimeCamelException rce = null; + try { + batchExchange = aggregate(rawMessages, session); + try { + processor.process(batchExchange); + } catch (Exception e) { + batchExchange.setException(e); + } + + // same evaluation as EndpointMessageListenerAsyncCallback.done(), sync branch only + if (batchExchange.isRollbackOnly()) { + rce = wrapRuntimeCamelException(new RollbackExchangeException(batchExchange)); + } else if (batchExchange.isFailed()) { + rce = wrapRuntimeCamelException(batchExchange.getException()); + } + } catch (Exception e) { + // aggregate() failed, so there is no exchange to inspect + rce = wrapRuntimeCamelException(e); + } finally { + if (batchExchange != null) { + // only after the outcome has been extracted + consumer.releaseExchange(batchExchange, false); + } + } + + if (rce != null) { + LOG.trace("onBatch END throwing exception: {}", rce.getMessage()); + throw rce; + } + } + + // BatchEndpointMessageListener Review Comment: This comment was just the file label in my snippet. It can go. ```suggestion ``` -- 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]
