Enable JDBC test cases with TestContainers approach

[mjd507]

There are few JDBC test cases currently are Disabled,
Now I want to enable them back with TestContainers approach like @artembilan mentioned MySqlContainerTest, OracleContainerTest, PostgresContainerTest

[mjd507]

hello @artembilan

I tried to enable DerbyTxTimeoutMessageStoreTests first. I didn't find the official TestContainer for Derby, so I use the Embedded one. we already have its dependency.

unluckily, I was stuck for few days for a test case testInt2993IdCacheConcurrency in its super class AbstractTxTimeoutMessageStoreTests,

the test starts 5 threads to concurrently pollMessageFromGroup and removeFromIdCache, both actions are in same transaction.

• Inside pollMessageFromGroup, it uses DerbyChannelMessageStoreQueryProvider to SELECT 1 message from database, add messageId to idCache(a HashSet), and DELETE the message from database.
• Inside removeFromIdCache, it uses idCacheWriteLock to ensure only one thread can remove messageId from idCache.

all looks good, but it failed every time in my local after running 60s :

java.sql.SQLTransactionRollbackException: A lock could not be obtained within the time requested

Possibly multiple transactions hold the same resource without releasing it, leading to lock timeout.

My confusion here is we only have one DELETE statement which could cause transactions queuing for resource competition. but it wouldn't take too much time, 60s is too long.

---

I create a new separate test class to analysis this, and found the thread hangs on this.idCacheWriteLock.lock(); in method removeFromIdCache .

this makes me more confuse. because by adding few logs around the writeLock, I can see every time writeLock is unlocked. why thread can not get a writeLock which is already unlocked?

below is the separate test class, we can copy and run in local to re-produce. I wonder is this only my machine issue?

many thanks

package org.springframework.integration.jdbc.store.channel;

import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.sql.DataSource;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.jupiter.api.Test;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
import org.springframework.integration.test.condition.LogLevels;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.jdbc.datasource.init.DataSourceInitializer;
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.support.TransactionTemplate;

import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType.DERBY;

@SpringJUnitConfig
@DirtiesContext
@LogLevels(categories = {
		"org.springframework.integration.jdbc",
//		"org.springframework.jdbc"
}, level = "TRACE")
public class DerbyTxTest {

	private static final Log log = LogFactory.getLog(DerbyTxTest.class);

	@Autowired
	JdbcChannelMessageStore jdbcChannelMessageStore;

	@Autowired
	PlatformTransactionManager transactionManager;

	@Test
	public void testConcurrentDelete() throws InterruptedException, ExecutionException {
		final String groupId = "testGroup";
		for (int i = 0; i < 5; i++) {
			this.jdbcChannelMessageStore.addMessageToGroup(groupId, new GenericMessage<String>("testMessage"));
		}

		ExecutorService executorService = Executors.newCachedThreadPool();
		CompletionService<Boolean> completionService = new ExecutorCompletionService<Boolean>(executorService);

		final int concurrency = 2;

		final TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
		transactionTemplate.setIsolationLevel(Isolation.READ_COMMITTED.value());

		for (int i = 0; i < concurrency; i++) {
			completionService.submit(() -> {
				for (int i1 = 0; i1 < 5; i1++) {
					boolean result = Boolean.TRUE.equals(transactionTemplate.execute(status -> {
						Message<?> message = null;
						try {
							message = jdbcChannelMessageStore.pollMessageFromGroup(groupId);
						}
						catch (Exception e1) {
							log.error("IdCache race condition.", e1);
							return false;
						}
						try {
							Thread.sleep(20);
						}
						catch (InterruptedException e2) {
							log.error(e2);
						}
						if (message != null) {
							// if I comment this line, test will pass
							jdbcChannelMessageStore.removeFromIdCache(message.getHeaders().getId().toString());
						}
						return true;
					}));
					if (!result) {
						return false;
					}
				}
				return true;
			});
		}

		for (int j = 0; j < concurrency; j++) {
			assertThat(completionService.take().get()).isTrue();
		}

		executorService.shutdown();
	}

	@Configuration
	@EnableIntegration
	public static class Config {

		@Bean
		DataSource dataSource() {
			return new EmbeddedDatabaseBuilder()
					.setType(DERBY)
					.build();
		}

		@Bean
		public DataSourceInitializer dataSourceInitializer(DataSource dataSource) {
			DataSourceInitializer initializer = new DataSourceInitializer();
			initializer.setDataSource(dataSource);

			ResourceDatabasePopulator populator = new ResourceDatabasePopulator();
			populator.addScript(new ClassPathResource("org/springframework/integration/jdbc/schema-derby.sql"));
			initializer.setDatabasePopulator(populator);
			initializer.setEnabled(true);

			return initializer;
		}

		@Bean
		PlatformTransactionManager transactionManager() {
			return new DataSourceTransactionManager(dataSource());
		}

		@Bean
		JdbcChannelMessageStore jdbcChannelMessageStore() {
			JdbcChannelMessageStore messageStore = new JdbcChannelMessageStore(dataSource());
			messageStore.setUsingIdCache(true);
			messageStore.setChannelMessageStoreQueryProvider(new DerbyChannelMessageStoreQueryProvider());
			return messageStore;
		}
	}
}

[artembilan]

I didn't find the official TestContainer for Derby, so I use the Embedded one. we already have its dependency.

That's correct. Why do we need a test container for such a simple embedded database?

I added this into the test:

	@BeforeAll
	static void setUp() {
		System.setProperty("derby.locks.deadlockTrace", "true");
	}

and got this in logs:

2025-05-27 12:02:21,218 ERROR [pool-2-thread-1] [org.springframework.integration.jdbc.store.channel.DerbyTxTest] - IdCache race condition.
org.springframework.dao.PessimisticLockingFailureException: PreparedStatementCallback; SQL [	SELECT INT_CHANNEL_MESSAGE.MESSAGE_ID, INT_CHANNEL_MESSAGE.MESSAGE_BYTES
	from INT_CHANNEL_MESSAGE
	where INT_CHANNEL_MESSAGE.GROUP_KEY = ? and INT_CHANNEL_MESSAGE.REGION = ?
and INT_CHANNEL_MESSAGE.MESSAGE_ID not in (?) order by CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY]; A lock could not be obtained within the time requested.  The lockTable dump is: 
Tue May 27 12:02:21 EDT 2025
XID       |TYPE         |MODE|LOCKCOUNT|LOCKNAME                                                                        |STATE|TABLETYPE / LOCKOBJ                   |INDEXNAME / CONTAINER_ID / (MODE for LATCH only)  |TABLENAME / CONGLOM_ID                |
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
*** The following row is the victim ***
251       |ROW          |S   |0        |(1,8)                                                                           |WAIT |T                                     |NULL                                              |INT_CHANNEL_MESSAGE                   |
*** The above row is the victim ***
247       |TABLE        |IX  |3        |Tablelock                                                                       |GRANT|T                                     |NULL                                              |INT_CHANNEL_MESSAGE                   |
251       |TABLE        |IS  |1        |Tablelock                                                                       |GRANT|T                                     |NULL                                              |INT_CHANNEL_MESSAGE                   |
247       |ROW          |X   |3        |(1,8)                                                                           |GRANT|T                                     |NULL                                              |INT_CHANNEL_MESSAGE                   |
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

I have no idea how to read this report, but it is indeed has something to do with in() operator in Derby.
I see that our transactions are committed.

Feels like no need to spend too much time on Derby.
Plus I see that this idCache algorithm is not OK: when we run concurrently there is high chance that all the thread are going into a state where they are able to to pull the same record from DB and all of them try to put this into a cache.

I would say FOR UPDATE variants are much better, then this not super useful idCache approach: https://docs.spring.io/spring-integration/reference/jdbc/message-store.html#concurrent-polling.

Different story, though.

Please, see if we can enable tests for enterprise RDBMS like Oracle, PostgreSQL and MySQL.
Thanks