gnodet-bot commented on code in PR #27173: URL: https://github.com/apache/camel/pull/27173#discussion_r4174018956
########## components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/BatchEndpointMessageListener.java: ########## @@ -0,0 +1,86 @@ +/* + * 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.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; + +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<Exchange>(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.getMessage().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; + try { + batchExchange = aggregate(rawMessages, session); + processor.process(batchExchange); + } catch (Exception e) { + batchExchange.setException(e); + } + + Exception exception = batchExchange.getException(); + + if (exception != null) { + LOG.trace("onBatch END throwing exception: {}", exception.getMessage()); + throw exception; + } Review Comment: 🔴 **Regression: `isRollbackOnly()` / `isFailed()` batches are now silently committed.** The previous version checked `batchExchange.isFailed() || batchExchange.isRollbackOnly()` and rolled back the session in those cases. This refactored version only throws (and thus triggers rollback in the caller) when `batchExchange.getException() != null`. If the route calls `.rollback()` or `.markRollbackOnly()` (which sets `isRollbackOnly()` without setting an exception), `onBatch()` returns normally → `BatchConsumerWorker.doOnBatch()` calls `commitIfNeeded()` → the batch is committed despite the route explicitly requesting rollback. Same for `exchange.setFailed(true)` without an exception — `isFailed()` is true but no exception is set. ```suggestion void onBatch(List<Message> rawMessages, Session session) throws Exception { LOG.trace("onBatch START"); LOG.debug("{} consumer received batch message: {}", endpoint, rawMessages); Exchange batchExchange = null; try { batchExchange = aggregate(rawMessages, session); processor.process(batchExchange); } catch (Exception e) { batchExchange.setException(e); } if (batchExchange.isFailed() || batchExchange.isRollbackOnly()) { Exception exception = batchExchange.getException(); String reason = exception != null ? exception.getMessage() : "rollbackOnly flag set"; LOG.trace("onBatch END throwing: {}", reason); throw exception != null ? exception : new org.apache.camel.CamelExchangeException("Batch rolled back (" + reason + ")", batchExchange); } } ``` ########## components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/BatchConsumerWorker.java: ########## @@ -0,0 +1,162 @@ +/* + * 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; + +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); + + BatchConsumerWorker(SjmsEndpoint endpoint, BatchEndpointMessageListener batchListener, + MessageConsumer consumer, Session session) { + this.endpoint = endpoint; + this.batchListener = batchListener; + this.consumer = consumer; + this.session = session; + } + + void shutdown() { + running.set(false); + } + + boolean isShutdownRequested() { + return !running.get(); + } + + private boolean isRedeliverable() { + return endpoint.isTransacted() + || endpoint.getAcknowledgementMode() == SessionAcknowledgementType.CLIENT_ACKNOWLEDGE; + } + + @Override + public void run() { + 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) { + onBatch(buffer); + buffer = new ArrayList<>(); + } + } + + if (!buffer.isEmpty()) { + onBatch(buffer); // graceful-stop drain + } + } catch (JMSException e) { + if (!buffer.isEmpty()) { + if (isRedeliverable()) { + LOG.error("Discarding {} buffered message(s) on {} after connection failure; " + + "unacknowledged/uncommitted, will be redelivered", + buffer.size(), endpoint.getEndpointUri()); + } else { + onBatch(buffer); + LOG.warn("Connection failed on {} with {} already-acknowledged message(s) buffered; " + + "attempting best-effort dispatch since they cannot be redelivered", + endpoint.getEndpointUri(), buffer.size()); + } + } + throw new BatchConsumerWorkerException(e); + } + } + + 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 { + LOG.warn("Execution of JMS message listener failed. This exception is ignored.", e); + } + } Review Comment: ⚠️ **Non-JMS route exceptions never reach the consumer's `ExceptionHandler`.** The non-batch `EndpointMessageListener` hands exceptions to `consumer.getExceptionHandler().handleException(...)`, which is what makes `bridgeErrorHandler` work — it replaces the default exception handler with one that propagates to the route's error handler. Here, non-JMS exceptions are just logged and swallowed (L131). This means `bridgeErrorHandler=true` has no effect on batch consumers. Please route to the consumer's exception handler: ```suggestion 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); } } batchListener.getConsumer().getExceptionHandler().handleException( "Error processing batch of " + batch.size() + " message(s)", e); } } ``` This requires exposing a `getConsumer()` accessor on `BatchEndpointMessageListener` (or passing the consumer reference directly to `BatchConsumerWorker`). ########## components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java: ########## @@ -449,6 +488,35 @@ public MessageListenerContainer createMessageListenerContainer(SjmsEndpoint endp return answer; } + public BatchMessageListenerContainer createBatchMessageListenerContainer( + SjmsEndpoint endpoint) { + BatchMessageListenerContainer answer = new BatchMessageListenerContainer(endpoint); + answer.setConcurrentConsumers(concurrentConsumers); + return answer; + } + + private void validateBatchingOptions() { + if (getExchangePattern().isOutCapable()) { + throw new IllegalArgumentException("SjmsConsumer does not support exchangePattern=InOut in batching mode"); + } + + if (getBatchInterval() < 0) { + if (getExchangePattern().isOutCapable()) { + throw new IllegalArgumentException("batchInterval must be 0 or greater."); + } Review Comment: 🔴 **Still dead code.** The `isOutCapable()` check at L499 already threw — so this inner `isOutCapable()` guard at L504 can never be true. Additionally, `batchSize <= 0` is still not validated; with `batchInterval=0` a batch never completes until the consumer stops. This was flagged in the previous 4 reviews by both gnodet-bot and @davsclaus. ```suggestion if (getBatchInterval() < 0) { throw new IllegalArgumentException("batchInterval must be 0 or greater."); } if (getBatchSize() <= 0) { throw new IllegalArgumentException("batchSize must be greater than 0."); } ``` ########## components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/BatchConsumerWorker.java: ########## @@ -0,0 +1,162 @@ +/* + * 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; + +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); + + BatchConsumerWorker(SjmsEndpoint endpoint, BatchEndpointMessageListener batchListener, + MessageConsumer consumer, Session session) { + this.endpoint = endpoint; + this.batchListener = batchListener; + this.consumer = consumer; + this.session = session; + } + + void shutdown() { + running.set(false); + } + + boolean isShutdownRequested() { + return !running.get(); + } + + private boolean isRedeliverable() { + return endpoint.isTransacted() + || endpoint.getAcknowledgementMode() == SessionAcknowledgementType.CLIENT_ACKNOWLEDGE; + } + + @Override + public void run() { + 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) { + onBatch(buffer); + buffer = new ArrayList<>(); + } + } + + if (!buffer.isEmpty()) { + onBatch(buffer); // graceful-stop drain Review Comment: ⚠️ **Connection loss: graceful-stop drain runs on a potentially dead session.** When `BatchMessageListenerContainer.onException(JMSException)` → `invalidateBatchWorkers()` → `worker.shutdown()` is called, the `while (running.get())` loop exits normally and hits this drain. The `doOnBatch` call processes the buffered messages through the route, then `commitIfNeeded` tries to commit on the (possibly dead) JMS session. With `transacted`/`CLIENT_ACKNOWLEDGE` the messages are redelivered by the broker after the dead session is cleaned up, so they get processed twice. Please discard the buffer when the shutdown is due to a connection failure (as the `JMSException` path at L106-116 already does), and only drain on a graceful application stop. One approach: add a `volatile boolean connectionLost` flag set by the container before calling `shutdown()`, and check it here. ########## components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConstants.java: ########## @@ -38,5 +38,8 @@ public interface SjmsConstants { description = "Provides an explicit ReplyTo destination (overrides any incoming value of Message.getJMSReplyTo() in consumer)", javaType = "String") String JMS_REPLY_TO = JmsConstants.JMS_REPLY_TO; - + @Metadata(label = "consumer, batch", + description = "The size of the batch when using the batching consumer option.", + javaType = "String") Review Comment: ⚠️ **Header metadata issues — still present.** Two problems: 1. `javaType = "String"` — should be `"Integer"` since the header value is an `int` (set via `rawMessages.size()`) 2. `label = "consumer, batch"` — the space after the comma produces `"group": " batch"` (with a leading space) in the generated JSON catalog ```suggestion @Metadata(label = "consumer,batch", description = "The size of the batch when using the batching consumer option.", javaType = "Integer") ``` ########## components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchTestSupport.java: ########## @@ -0,0 +1,74 @@ +/* + * 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.List; + +import org.apache.camel.Exchange; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.sjms.SjmsConstants; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +@SuppressWarnings("unchecked") +public final class BatchTestSupport { + + private BatchTestSupport() { + } + + public static void sendMessages(ProducerTemplate template, String endpoint, int count) { + for (int i = 0; i < count; i++) { + String message = "Hello World " + i; + template.sendBody(endpoint, message); + } + } + + /** Extracts the batch body as a List<Exchange> from a batch Exchange, asserting the type. */ + public static List<Exchange> getBatch(Exchange batchExchange) { + List<Exchange> body = batchExchange.getIn().getBody(List.class); + assertNotNull(body, "batch exchange body was null"); + for (Object o : body) { + assertInstanceOf(Exchange.class, o, "batch element was not an Exchange: " + o); + } + return body; + } + + /** Asserts a single batch exchange has exactly the given number of messages. */ + public static void assertBatchSize(Exchange batchExchange, int expectedSize) { + List<Exchange> batch = getBatch(batchExchange); + assertEquals(expectedSize, batch.size(), "unexpected batch size"); + assertEquals(expectedSize, + batchExchange.getIn().getHeader(SjmsConstants.SJMS_BATCH_SIZE_HEADER, Integer.class), + "CamelSjmsBatchSize header did not match actual batch size"); + } + + /** + * Asserts the mock received exactly expectedSizes.length batch exchanges, in order, with each batch's size matching + * the corresponding element. Requires concurrentConsumers=1 (or an otherwise deterministic single-worker setup) so + * that arrival order is meaningful. + */ + public static void assertBatchSizesInOrder(MockEndpoint mock, int... expectedSizes) { + assertEquals(mock.getExchanges().size(), expectedSizes.length, "More mock exchanges than provided batch sizes."); Review Comment: ⚠️ **`assertEquals` arguments still swapped.** JUnit 5's signature is `assertEquals(expected, actual, message)`. Here `mock.getExchanges().size()` (the actual value) is in the expected position. This was flagged in every previous review. ```suggestion assertEquals(expectedSizes.length, mock.getExchanges().size(), "More mock exchanges than provided batch sizes."); ``` ########## components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/BatchEndpointMessageListener.java: ########## @@ -0,0 +1,86 @@ +/* + * 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.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; + +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<Exchange>(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.getMessage().setHeader(SjmsConstants.SJMS_BATCH_SIZE_HEADER, rawMessages.size()); Review Comment: 💡 **Inconsistent API usage — still present.** `setBody()` uses `getIn()` (line 60) while `setHeader()` uses `getMessage()` (line 62). In Camel 4.x these return the same object, but mixing them is confusing. Use `getIn().setHeader(...)` for consistency with the rest of the camel-sjms codebase. -- 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]
