spring-projectsGH-8680: Check DB for table on start

Fixes spring-projects#8680

If database is not initialized properly before application start, we may lose messages
at runtime when we fail to insert data into DB

* Implement `SmartLifecycle` on `JdbcMessageStore`, `JdbcChannelMessageStore`,
`JdbcMetadataStore`, and `DefaultLockRepository` to perform `SELECT COUNT` query in `start()`
to fail fast if no required table is present.
* Refactor `AbstractJdbcChannelMessageStoreTests` into JUnit 5 and use `MySqlContainerTest`
for more coverage
* Fix newly failed tests which had DB not initialized before
* Exclude `commons-logging` from `commons-dbcp2` dependency to avoid
classpath conflict
* Document the new feature

Author: artembilan

build.gradle

@@ -740,7 +740,9 @@ project('spring-integration-jdbc') {
         testImplementation "org.apache.derby:derbyclient:$derbyVersion"
         testImplementation "org.postgresql:postgresql:$postgresVersion"
         testImplementation "mysql:mysql-connector-java:$mysqlVersion"
-        testImplementation "org.apache.commons:commons-dbcp2:$commonsDbcp2Version"
+        testImplementation ("org.apache.commons:commons-dbcp2:$commonsDbcp2Version") {
+            exclude group: 'commons-logging'
+        }
         testImplementation 'org.testcontainers:mysql'
         testImplementation 'org.testcontainers:postgresql'
 

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java

@@ -20,6 +20,7 @@
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.sql.DataSource;
 
@@ -29,6 +30,7 @@
 import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.transaction.PlatformTransactionManager;
@@ -45,6 +47,11 @@
  * Otherwise, it opens a possibility to break {@link java.util.concurrent.locks.Lock} contract,
  * where {@link JdbcLockRegistry} uses non-shared {@link java.util.concurrent.locks.ReentrantLock}s
  * for local synchronizations.
+ * <p>
+ * This class implements {@link SmartLifecycle} and calls
+ * {@code SELECT COUNT(REGION) FROM %sLOCK} query
+ * according to the provided prefix on {@link #start()} to check if required table is present in DB.
+ * The application context is going to fail starting if table is not present.
  *
  * @author Dave Syer
  * @author Artem Bilan
@@ -56,7 +63,8 @@
  * @since 4.3
  */
 public class DefaultLockRepository
-		implements LockRepository, InitializingBean, ApplicationContextAware, SmartInitializingSingleton {
+		implements LockRepository, InitializingBean, ApplicationContextAware, SmartInitializingSingleton,
+		SmartLifecycle {
 
 	/**
 	 * Default value for the table prefix property.
@@ -72,6 +80,8 @@ public class DefaultLockRepository
 
 	private final JdbcTemplate template;
 
+	private final AtomicBoolean started = new AtomicBoolean();
+
 	private Duration ttl = DEFAULT_TTL;
 
 	private String prefix = DEFAULT_TABLE_PREFIX;
@@ -116,6 +126,10 @@ SELECT COUNT(REGION)
 			WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=?
 			""";
 
+	private String countAllQuery = """
+			SELECT COUNT(REGION) FROM %sLOCK
+			""";
+
 	private ApplicationContext applicationContext;
 
 	private PlatformTransactionManager transactionManager;
@@ -293,6 +307,7 @@ public void afterPropertiesSet() {
 		this.insertQuery = String.format(this.insertQuery, this.prefix);
 		this.countQuery = String.format(this.countQuery, this.prefix);
 		this.renewQuery = String.format(this.renewQuery, this.prefix);
+		this.countAllQuery = String.format(this.countAllQuery, this.prefix);
 	}
 
 	@Override
@@ -325,6 +340,23 @@ public void afterSingletonsInstantiated() {
 		this.serializableTransactionTemplate = new TransactionTemplate(this.transactionManager, transactionDefinition);
 	}
 
+	@Override
+	public void start() {
+		if (this.started.compareAndSet(false, true)) {
+			this.template.queryForObject(this.countAllQuery, Integer.class); // If no table in DB, an exception is thrown
+		}
+	}
+
+	@Override
+	public void stop() {
+		this.started.set(false);
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.started.get();
+	}
+
 	@Override
 	public void close() {
 		this.defaultTransactionTemplate.executeWithoutResult(

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStore.java

@@ -16,9 +16,12 @@
 
 package org.springframework.integration.jdbc.metadata;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import javax.sql.DataSource;
 
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.integration.metadata.ConcurrentMetadataStore;
@@ -34,14 +37,19 @@
  * where <code>*</code> is the target database type.
  * <p>
  * The transaction management is required to use this {@link ConcurrentMetadataStore}.
+ * <p>
+ * This class implements {@link SmartLifecycle} and calls
+ * {@code SELECT COUNT(METADATA_KEY) FROM %sMETADATA_STORE} query
+ * according to the provided prefix on {@link #start()} to check if required table is present in DB.
+ * The application context is going to fail starting if table is not present.
  *
  * @author Bojan Vukasovic
  * @author Artem Bilan
  * @author Gary Russell
  *
  * @since 5.0
  */
-public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingBean {
+public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingBean, SmartLifecycle {
 
 	private static final String KEY_CANNOT_BE_NULL = "'key' cannot be null";
 
@@ -52,6 +60,8 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
 
 	private final JdbcOperations jdbcTemplate;
 
+	private final AtomicBoolean started = new AtomicBoolean();
+
 	private String tablePrefix = DEFAULT_TABLE_PREFIX;
 
 	private String region = "DEFAULT";
@@ -93,6 +103,10 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
 			HAVING COUNT(*)=0
 			""";
 
+	private String countQuery = """
+			SELECT COUNT(METADATA_KEY) FROM %sMETADATA_STORE
+			""";
+
 	/**
 	 * Instantiate a {@link JdbcMetadataStore} using provided dataSource {@link DataSource}.
 	 * @param dataSource a {@link DataSource}
@@ -137,7 +151,7 @@ public void setRegion(String region) {
 	 * Specify a row lock hint for the query in the lock-based operations.
 	 * Defaults to {@code FOR UPDATE}. Can be specified as an empty string,
 	 * if the target RDBMS doesn't support locking on tables from queries.
-	 * The value depends from RDBMS vendor, e.g. SQL Server requires {@code WITH (ROWLOCK)}.
+	 * The value depends on RDBMS vendor, e.g. SQL Server requires {@code WITH (ROWLOCK)}.
 	 * @param lockHint the RDBMS vendor-specific lock hint.
 	 * @since 5.0.7
 	 */
@@ -154,6 +168,24 @@ public void afterPropertiesSet() {
 		this.replaceValueByKeyQuery = String.format(this.replaceValueByKeyQuery, this.tablePrefix);
 		this.removeValueQuery = String.format(this.removeValueQuery, this.tablePrefix);
 		this.putIfAbsentValueQuery = String.format(this.putIfAbsentValueQuery, this.tablePrefix, this.tablePrefix);
+		this.countQuery = String.format(this.putIfAbsentValueQuery, this.tablePrefix);
+	}
+
+	@Override
+	public void start() {
+		if (this.started.compareAndSet(false, true)) {
+			this.jdbcTemplate.queryForObject(this.countQuery, Integer.class); // If no table in DB, an exception is thrown
+		}
+	}
+
+	@Override
+	public void stop() {
+		this.started.set(false);
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.started.get();
 	}
 
 	@Override
@@ -162,7 +194,7 @@ public String putIfAbsent(String key, String value) {
 		Assert.notNull(key, KEY_CANNOT_BE_NULL);
 		Assert.notNull(value, "'value' cannot be null");
 		while (true) {
-			//try to insert if does not exists
+			//try to insert if entry does not exist
 			int affectedRows = tryToPutIfAbsent(key, value);
 			if (affectedRows > 0) {
 				//it was not in the table, so we have just inserted
@@ -218,7 +250,7 @@ public void put(String key, String value) {
 		Assert.notNull(key, KEY_CANNOT_BE_NULL);
 		Assert.notNull(value, "'value' cannot be null");
 		while (true) {
-			//try to insert if does not exist, if exists we will try to update it
+			//try to insert if entry does not exist, if exists we will try to update it
 			int affectedRows = tryToPutIfAbsent(key, value);
 			if (affectedRows == 0) {
 				//since value is not inserted, means it is already present

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -31,6 +32,7 @@
 import javax.sql.DataSource;
 
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.core.log.LogMessage;
 import org.springframework.core.serializer.Deserializer;
@@ -73,6 +75,10 @@
  * The SQL scripts for creating the table are packaged
  * under {@code org/springframework/integration/jdbc/schema-*.sql},
  * where {@code *} denotes the target database type.
+ * <p>
+ * This class implements {@link SmartLifecycle} and calls {@link #getMessageGroupCount()}
+ * on {@link #start()} to check if required table is present in DB.
+ * The application context is going to fail starting if table is not present.
  *
  * @author Gunnar Hillert
  * @author Artem Bilan
@@ -83,7 +89,7 @@
  * @since 2.2
  */
 @ManagedResource
-public class JdbcChannelMessageStore implements PriorityCapableChannelMessageStore, InitializingBean {
+public class JdbcChannelMessageStore implements PriorityCapableChannelMessageStore, InitializingBean, SmartLifecycle {
 
 	private static final LogAccessor LOGGER = new LogAccessor(JdbcChannelMessageStore.class);
 
@@ -121,6 +127,8 @@ private enum Query {
 
 	private final Lock idCacheWriteLock = this.idCacheLock.writeLock();
 
+	private final AtomicBoolean started = new AtomicBoolean();
+
 	private ChannelMessageStoreQueryProvider channelMessageStoreQueryProvider;
 
 	private String region = DEFAULT_REGION;
@@ -411,6 +419,23 @@ public void afterPropertiesSet() {
 		this.jdbcTemplate.afterPropertiesSet();
 	}
 
+	@Override
+	public void start() {
+		if (this.started.compareAndSet(false, true)) {
+			getMessageGroupCount(); // If no table in DB, an exception is thrown
+		}
+	}
+
+	@Override
+	public void stop() {
+		this.started.set(false);
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.started.get();
+	}
+
 	/**
 	 * Store a message in the database. The groupId identifies the channel for which
 	 * the message is to be stored.

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java

@@ -26,11 +26,13 @@
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Stream;
 
 import javax.sql.DataSource;
 
 import org.springframework.beans.factory.BeanClassLoaderAware;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
 import org.springframework.core.serializer.support.SerializingConverter;
@@ -66,6 +68,10 @@
  * please consider using the channel-specific {@link JdbcChannelMessageStore} instead.
  * This implementation is intended for correlation components (e.g. {@code <aggregator>}),
  * {@code <delayer>} and similar.
+ * <p>
+ * This class implements {@link SmartLifecycle} and calls {@link #getMessageGroupCount()}
+ * on {@link #start()} to check if required tables are present in DB.
+ * The application context is going to fail starting if tables are not present.
  *
  * @author Dave Syer
  * @author Oleg Zhurakousky
@@ -77,7 +83,8 @@
  *
  * @since 2.0
  */
-public class JdbcMessageStore extends AbstractMessageGroupStore implements MessageStore, BeanClassLoaderAware {
+public class JdbcMessageStore extends AbstractMessageGroupStore
+		implements MessageStore, BeanClassLoaderAware, SmartLifecycle {
 
 	/**
 	 * Default value for the table prefix property.
@@ -234,6 +241,8 @@ public String getSql() {
 
 	private final Map<Query, String> queryCache = new ConcurrentHashMap<>();
 
+	private final AtomicBoolean started = new AtomicBoolean();
+
 	private String region = "DEFAULT";
 
 	private String tablePrefix = DEFAULT_TABLE_PREFIX;
@@ -331,6 +340,23 @@ public void addAllowedPatterns(String... patterns) {
 		this.deserializer.addAllowedPatterns(patterns);
 	}
 
+	@Override
+	public void start() {
+		if (this.started.compareAndSet(false, true)) {
+			getMessageGroupCount(); // If no table in DB, an exception is thrown
+		}
+	}
+
+	@Override
+	public void stop() {
+		this.started.set(false);
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.started.get();
+	}
+
 	@Override
 	public Message<?> removeMessage(UUID id) {
 		Message<?> message = getMessage(id);

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,8 +20,8 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 
-import org.junit.After;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.core.serializer.DefaultDeserializer;
@@ -77,11 +77,11 @@ public void testMessageStoreWithAttributes() {
 		setUp("soupedUpJdbcMessageStore.xml", getClass());
 		MessageStore store = context.getBean("messageStore", MessageStore.class);
 		assertThat(ReflectionTestUtils.getField(store, "region")).isEqualTo("FOO");
-		assertThat(ReflectionTestUtils.getField(store, "tablePrefix")).isEqualTo("BAR_");
+		assertThat(ReflectionTestUtils.getField(store, "tablePrefix")).isEqualTo("int_");
 		assertThat(ReflectionTestUtils.getField(store, "lobHandler")).isEqualTo(context.getBean(LobHandler.class));
 	}
 
-	@After
+	@AfterEach
 	public void tearDown() {
 		if (context != null) {
 			context.close();