gnodet-bot commented on code in PR #27173:
URL: https://github.com/apache/camel/pull/27173#discussion_r4169785873
##########
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:
🔴 **Bug: negative `batchInterval` validation is dead code.** The
`isOutCapable()` check at line 499 already throws before reaching this block —
so the inner `isOutCapable()` guard at line 504 can never be true (it's already
handled). The intent was to reject negative `batchInterval` values regardless
of exchange pattern.
```suggestion
if (getBatchInterval() < 0) {
throw new IllegalArgumentException("batchInterval must be 0 or
greater.");
}
```
##########
components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/BatchEndpointMessageListener.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.apache.camel.component.sjms.SjmsHelper;
+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 {
+ Exchange batchExchange = null;
+ Exception failure = null;
+ try {
+ batchExchange = aggregate(rawMessages, session);
+ processor.process(batchExchange);
+ } catch (Exception e) {
+ failure = e;
+ }
+
+ boolean failed = failure != null || batchExchange.isFailed() ||
batchExchange.isRollbackOnly();
+ Message lastMessage = rawMessages.get(rawMessages.size() - 1);
+
+ if (!failed) {
+ SjmsHelper.commitIfNeeded(session, lastMessage);
+ } else {
+ Exception cause = failure != null ? failure :
batchExchange.getException();
+ if (cause != null) {
+ LOG.warn("Batch of {} message(s) failed processing on {}: {}",
rawMessages.size(),
+ endpoint.getEndpointUri(), cause.getMessage(), cause);
+ }
Review Comment:
⚠️ **Silent failure when `isRollbackOnly()` without exception.** If
`batchExchange.isRollbackOnly()` is set without an exception (e.g. by a
`.rollback()` DSL step), both `failure` and `batchExchange.getException()` are
null, so `cause` is null and the `LOG.warn` is skipped. The session is rolled
back correctly, but there's no log trace explaining *why*.
```suggestion
Exception cause = failure != null ? failure :
batchExchange.getException();
if (cause != null) {
LOG.warn("Batch of {} message(s) failed processing on {}:
{}", rawMessages.size(),
endpoint.getEndpointUri(), cause.getMessage(),
cause);
} else {
LOG.warn("Batch of {} message(s) rolled back on {}
(rollbackOnly flag set)", rawMessages.size(),
endpoint.getEndpointUri());
}
```
##########
components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/BatchEndpointMessageListener.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.apache.camel.component.sjms.SjmsHelper;
+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:** `setBody()` uses `getIn()` (line 60) while
`setHeader()` uses `getMessage()` (line 63). In Camel 4.x these return the same
object, but mixing them is confusing and inconsistent with the rest of the
camel-sjms codebase which uniformly uses `getIn()`. Use
`getIn().setHeader(...)` for consistency.
##########
components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchTestSupport.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.*;
Review Comment:
⚠️ **Star import:** `org.apache.camel.component.sjms.*` — the project uses
explicit imports. This was the same finding as #1 from the original review on
`SjmsEndpoint.java`.
```suggestion
import org.apache.camel.component.sjms.SjmsConstants;
```
(The `BatchEndpointMessageListener` import on line 25 is unused too —
`BatchTestSupport` doesn't reference it.)
##########
components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchTestSupport.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.*;
+import org.apache.camel.component.sjms.consumer.BatchEndpointMessageListener;
+
+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 swapped.** JUnit 5's signature is
`assertEquals(expected, actual, message)`. Here `mock.getExchanges().size()`
(the actual value) is in the expected position. This was already flagged by
@davsclaus in finding #12.
```suggestion
assertEquals(expectedSizes.length, mock.getExchanges().size(), "More
mock exchanges than provided batch sizes.");
```
--
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]
