gnodet-bot commented on code in PR #27173:
URL: https://github.com/apache/camel/pull/27173#discussion_r4171983799
##########
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 negative `batchInterval` check at L503-506 is
still nested inside `isOutCapable()` (L499), which already threw at L500. This
block can never execute. Additionally, `batchSize <= 0` is still not validated.
As flagged by @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/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:
⚠️ **Still silent on rollback-only.** When `batchExchange.isRollbackOnly()`
is set without an exception (e.g. via `.rollback()` DSL), both `failure` and
`batchExchange.getException()` are null, so `cause` is null and the `LOG.warn`
is skipped. The session is rolled back silently with no trace.
```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/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.*;
Review Comment:
⚠️ **Star import still present.** `org.apache.camel.component.sjms.*` — the
project uses explicit imports. The unused `BatchEndpointMessageListener` import
was removed in this commit, but the star import remains.
```suggestion
import org.apache.camel.component.sjms.SjmsConstants;
```
##########
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",
Review Comment:
⚠️ **Header metadata issues.** Two problems with this annotation:
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.*;
+
+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 signature is
`assertEquals(expected, actual, message)`. Here `mock.getExchanges().size()`
(the actual) is in the expected position.
```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]
