davsclaus commented on code in PR #27173: URL: https://github.com/apache/camel/pull/27173#discussion_r4335773714
########## components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerRollbackOnlyTest.java: ########## @@ -0,0 +1,134 @@ +/* + * 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.Exchange; +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.assertBatchRedelivered; +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.sendMessages; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.sendMessagesWithText; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class BatchConsumerRollbackOnlyTest extends JmsTestSupport { + + private static final String QUEUE_NAME_TEMPLATE = "batch.consumer.%s.BatchConsumerRollbackOnlyTest"; + private static final String ROUTE_ID_AUTO_ACK = "auto-ack"; + private static final String ROUTE_ID_SESSION_TX = "session-tx"; + private static final String ROUTE_ID_CLIENT_ACK = "client-ack"; + private static final String MESSAGE_TEXT = "Message %d"; + + @Test + public void testRollbackAutoAcknowledge() throws Exception { + MockEndpoint mockStart = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_START, ROUTE_ID_AUTO_ACK)); + mockStart.expectedMessageCount(1); + + MockEndpoint mockFinish = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_FINISH, ROUTE_ID_AUTO_ACK)); + // With AUTO_ACKNOWLEDGE the messages in the batch will not be redelivered and reach end of the route + mockFinish.expectedMessageCount(0); + + sendMessages(template, format("sjms:queue:" + QUEUE_NAME_TEMPLATE, ROUTE_ID_AUTO_ACK), + 5); + + MockEndpoint.assertIsSatisfied(context); + assertBatchSizesInOrder(mockStart, 5); + } + + @Test + public void testRollbackSessionTransacted() 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)); + // With session transacted the messages in the batch will be redelivered and Review Comment: Same unfinished sentence as in `BatchConsumerTransactedTest`: ```suggestion // With session transacted the messages in the batch will be redelivered and reach end of the route ``` ########## components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerTransactedTest.java: ########## @@ -0,0 +1,132 @@ +/* + * 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.Exchange; +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.assertAllBatchMessagesRedelivered; +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 = "session-tx"; + private static final String ROUTE_ID_CLIENT_ACK = "client-ack"; + private static final String ROUTE_ID_AUTO_ACK = "auto-ack"; + private static final String MESSAGE_TEXT = "Message %d"; + + @Test + public void testClientAcknowledged() throws Exception { + MockEndpoint mockStart = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_START, ROUTE_ID_CLIENT_ACK)); + mockStart.expectedMessageCount(2); + + MockEndpoint mockFinish = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_FINISH, ROUTE_ID_CLIENT_ACK)); + // With CLIENT_ACKNOWLEDGE the messages in the batch will be redelivered and reach end of the route + mockFinish.expectedMessageCount(1); + + sendMessagesWithText(template, + format("sjms:queue:" + QUEUE_NAME_TEMPLATE, ROUTE_ID_CLIENT_ACK), + 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))); + assertAllBatchMessagesRedelivered(mockFinish.getExchanges().get(0)); + } + + @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)); + // With session transacted the messages in the batch will be redelivered and Review Comment: The sentence stops after "and". Hint: finish it like the `CLIENT_ACKNOWLEDGE` one above: ```suggestion // With session transacted the messages in the batch will be redelivered and reach end of the route ``` ########## components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerExceptionHandlerTest.java: ########## @@ -0,0 +1,107 @@ +/* + * 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.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.camel.BindToRegistry; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.sjms.support.JmsTestSupport; +import org.apache.camel.spi.ExceptionHandler; +import org.junit.jupiter.api.Test; + +import static org.apache.camel.component.sjms.batch.BatchTestHelper.sendMessagesWithText; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +public class BatchConsumerExceptionHandlerTest extends JmsTestSupport { + + private static final String SJMS_FROMF_URI = "%s?batching=true&batchSize=5&exceptionHandler=#myHandler"; + private static final String SJMS_QUEUE_NAME = "sjms:queue:batch.consumer.BatchConsumerExceptionHandlerTest"; + private static final String MOCK_START = "mock:start"; + private static final String MOCK_FINISH = "mock:finish"; + private static final String EXCEPTION_MESSAGE = "Always throw an exception!"; + + @BindToRegistry("myHandler") + private CapturingExceptionHandler exceptionHandler = new CapturingExceptionHandler(); + + @Test + public void testBatchConsumerExceptionHandler() throws InterruptedException { + MockEndpoint mockStart = getMockEndpoint(MOCK_START); + MockEndpoint mockFinish = getMockEndpoint(MOCK_FINISH); + + mockStart.expectedMessageCount(1); + + mockFinish.expectedMessageCount(0); + + sendMessagesWithText(template, SJMS_QUEUE_NAME, 5, "Message %d!"); + + await() + .atMost(Duration.ofSeconds(10)) + .until(() -> exceptionHandler.counter.get() == 1); Review Comment: oscerd suggested keeping one of the two waits; now both `setAssertPeriod(5000)` and `.during(5s)` are gone. The `until` returns as soon as the first call lands, so `assertEquals(1, exceptionHandler.counter.get())` no longer shows that there is no second call. Hint: keep a short `during`, for example: ```java await() .during(Duration.ofSeconds(2)) .atMost(Duration.ofSeconds(10)) .until(() -> exceptionHandler.counter.get() == 1); ``` ########## components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerStopStartTest.java: ########## @@ -0,0 +1,235 @@ +/* + * 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.time.Duration; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.CamelContext; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.sjms.support.JmsExclusiveTestSupport; +import org.apache.camel.test.infra.artemis.services.ArtemisService; +import org.apache.camel.test.infra.artemis.services.ArtemisServiceFactory; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +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.assertBatchSizesInOrder; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.countWorkerThreads; +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.sendMessages; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.sendMessagesWithText; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class BatchConsumerStopStartTest extends JmsExclusiveTestSupport { + + private static final String QUEUE_NAME_TEMPLATE = "batch.consumer.%s.BatchConsumerStopStartTest"; + private static final String ROUTE_ID_SESSION_TX = "tx"; + private static final String ROUTE_ID_CLIENT_ACK = "client-ack"; + private static final String ROUTE_ID_AUTO_ACK = "auto-ack"; + private static final String SJMS_CONNECTION_FACTORY = "countingConsumerConnectionFactory"; + + private BatchTestHelper.TrackedConnectionFactory trackedConnectionFactory; + + @RegisterExtension + public static ArtemisService service = ArtemisServiceFactory.createVMService(); + + @Override + public ArtemisService getService() { + return service; + } + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext camelContext = super.createCamelContext(); + + trackedConnectionFactory = new BatchTestHelper.TrackedConnectionFactory(connectionFactory); + camelContext.getRegistry().bind(SJMS_CONNECTION_FACTORY, trackedConnectionFactory.proxy()); + + return camelContext; + } + + @Test + public void testAutoAcknowledgePartialBatchDrainedOnRouteStop() throws Exception { + final String queueName = format(QUEUE_NAME_TEMPLATE, ROUTE_ID_AUTO_ACK); + final String endpointUri = "sjms:queue:" + queueName; + + MockEndpoint mock = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_FINISH, ROUTE_ID_AUTO_ACK)); + + // An incomplete batch of AUTO_ACKNOWLEDGE messages should be processed on route stop + mock.expectedMessageCount(1); + + sendMessagesWithText(template, endpointUri, 3, "Before!"); + + await().atMost(Duration.ofSeconds(5)) + .until(() -> trackedConnectionFactory.messages(queueName) == 3); + + context.getRouteController().stopRoute(ROUTE_ID_AUTO_ACK); + + mock.assertIsSatisfied(); + assertEquals(List.of("Before!", "Before!", "Before!"), getBatchBodiesAsString(mock.getExchanges().get(0))); + + // AUTO_ACKNOWLEDGE messages processed in the incomplete batch should no longer be in the queue and therefore not appear in the batch after restart + mock.reset(); + mock.expectedMessageCount(1); + + context.getRouteController().startRoute(ROUTE_ID_AUTO_ACK); + sendMessagesWithText(template, endpointUri, 5, "After!"); + + mock.assertIsSatisfied(); + assertEquals(List.of("After!", "After!", "After!", "After!", "After!"), + getBatchBodiesAsString(mock.getExchanges().get(0))); + } + + @Test + public void testSessionTransactedPartialBatchNotDrainedOnRouteStop() throws Exception { + final String queueName = format(QUEUE_NAME_TEMPLATE, ROUTE_ID_SESSION_TX); + final String endpointUri = "sjms:queue:" + queueName; + + MockEndpoint mock = getMockEndpoint(format(BATCH_ROUTEBUILDER_MOCK_FINISH, ROUTE_ID_SESSION_TX)); + + // An incomplete batch os SESSION_TRANSACTED messages should not be processed on route stop Review Comment: Typo from oscerd's earlier note ("os" for "of"): ```suggestion // An incomplete batch of SESSION_TRANSACTED messages should not be processed on route stop ``` -- 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]
