davsclaus commented on code in PR #27173: URL: https://github.com/apache/camel/pull/27173#discussion_r4357362591
########## components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerRecoveryFailedAttemptTest.java: ########## @@ -0,0 +1,231 @@ +/* + * 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.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import jakarta.jms.Connection; +import jakarta.jms.ConnectionFactory; +import jakarta.jms.ExceptionListener; +import jakarta.jms.JMSContext; +import jakarta.jms.JMSException; +import jakarta.jms.Session; + +import org.apache.camel.BindToRegistry; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.sjms.support.JmsTestSupport; +import org.junit.jupiter.api.Test; + +import static org.apache.camel.component.sjms.batch.BatchTestHelper.countWorkerThreads; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.receivedMessages; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.sendMessagesWithText; +import static org.apache.camel.component.sjms.batch.BatchTestHelper.triggerConnectionFailure; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests the connection recovery of the batch consumer. The recovery creates a new connection and then the sessions and + * consumers on it. This test covers the case where the new connection fails when it is started, that is before the + * sessions and consumers are created, and the recovery has to try again with another connection. The connections, + * sessions and consumers of the route are tracked through a connection factory that can make the next connection fail + * when it is started. + */ +class BatchConsumerRecoveryFailedAttemptTest extends JmsTestSupport { + + private static final String QUEUE_NAME = "BatchConsumerRecoveryFailedAttemptTest"; + private static final String SJMS_FROMF_URI + = "sjms:queue:%s?batching=true&batchSize=5&concurrentConsumers=%d&recoveryInterval=200&connectionFactory=#trackingConnectionFactory"; + private static final int CONSUMERS = 3; + + private final List<TrackedConnection> connections = new CopyOnWriteArrayList<>(); + // whether the next connection created by the tracking connection factory fails when it is started; + // the connection resets the flag when it takes it + private volatile boolean nextFailsOnStart; + + @BindToRegistry("trackingConnectionFactory") + private final ConnectionFactory trackingConnectionFactory = new ConnectionFactory() { + @Override + public Connection createConnection() throws JMSException { + return track(connectionFactory.createConnection()); + } + + @Override + public Connection createConnection(String userName, String password) throws JMSException { + return track(connectionFactory.createConnection(userName, password)); + } + + @Override + public JMSContext createContext() { + return connectionFactory.createContext(); + } + + @Override + public JMSContext createContext(String userName, String password) { + return connectionFactory.createContext(userName, password); + } + + @Override + public JMSContext createContext(String userName, String password, int sessionMode) { + return connectionFactory.createContext(userName, password, sessionMode); + } + + @Override + public JMSContext createContext(int sessionMode) { + return connectionFactory.createContext(sessionMode); + } + }; + + /** + * The new connection reports its failure to its exception listener when the recovery starts it, before the + * consumers are created, and it delivers no message. The JMS provider still creates the sessions and consumers on + * it. The recovery has to notice the failure and try again with another connection, although onException found no + * consumers to drop and none of the create calls failed. Afterwards the failed connection is closed, the connection + * in use has one session and one consumer per concurrent consumer, and the route processes messages again. + */ + @Test + void testRecoveryTriesAgainWhenConnectionFailsBeforeConsumersAreCreated() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:beforeconsumers"); + mock.setMinimumExpectedMessageCount(4); + + sendMessagesWithText(template, "sjms:queue:" + QUEUE_NAME, 20, "Before!"); + + mock.assertIsSatisfied(); + assertEquals(20, receivedMessages(mock)); Review Comment: This returns as soon as the 4th batch arrives (`setMinimumExpectedMessageCount(4)`). With batches of 5 spread over 3 consumers, the 20 messages usually come as three full batches plus up to three partial ones. The 1-second `batchInterval` completes the partial ones a few ms apart, so this sum can still be 16 or 18 here. The same applies at line 138. Hint: wait for the count, as `BatchConcurrentConsumerTest` does: ```suggestion await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(20, receivedMessages(mock))); ``` -- 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]
