INT-4499: Add JdbcMetadataStore.setLockHint()

JIRA https://jira.spring.io/browse/INT-4499
Fixes #2483

To allow to reconfigure a `SELECT ... FOR UPDATE` for particular RDBMS
vendor, we need an option to specify possible hint for the target data base.

**Cherry-pick to 5.0.x**

Author: artembilan

spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -58,7 +58,7 @@ public class PersistentAcceptOnceFileListFilterExternalStoreTests extends RedisA
 	@Test
 	@RedisAvailable
 	public void testFileSystemWithRedisMetadataStore() throws Exception {
-		RedisTemplate<String, ?> template = new RedisTemplate<String, Object>();
+		RedisTemplate<String, ?> template = new RedisTemplate<>();
 		template.setConnectionFactory(this.getConnectionFactoryForTest());
 		template.setKeySerializer(new StringRedisSerializer());
 		template.afterPropertiesSet();
@@ -87,6 +87,7 @@ public void testFileSystemWithJdbcMetadataStore() throws Exception {
 				.build();
 
 		JdbcMetadataStore metadataStore = new JdbcMetadataStore(dataSource);
+		metadataStore.setLockHint("");
 		metadataStore.afterPropertiesSet();
 
 		try {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -49,32 +49,24 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
 
 	private final JdbcOperations jdbcTemplate;
 
-	private volatile String tablePrefix = DEFAULT_TABLE_PREFIX;
+	private String tablePrefix = DEFAULT_TABLE_PREFIX;
 
-	private volatile String region = "DEFAULT";
+	private String region = "DEFAULT";
 
-	private String getValueQuery = "SELECT METADATA_VALUE FROM %SMETADATA_STORE WHERE METADATA_KEY=? AND REGION=?";
+	private String lockHint = "FOR UPDATE";
 
-	private String getValueForUpdateQuery = "SELECT METADATA_VALUE FROM %SMETADATA_STORE WHERE METADATA_KEY=? AND REGION=? FOR UPDATE";
+	private String getValueQuery = "SELECT METADATA_VALUE FROM %sMETADATA_STORE WHERE METADATA_KEY=? AND REGION=?";
 
-	private String replaceValueQuery = "UPDATE %SMETADATA_STORE SET METADATA_VALUE=? WHERE METADATA_KEY=? AND METADATA_VALUE=? AND REGION=?";
+	private String getValueForUpdateQuery = "SELECT METADATA_VALUE FROM %sMETADATA_STORE WHERE METADATA_KEY=? AND REGION=? %s";
 
-	private String replaceValueByKeyQuery = "UPDATE %SMETADATA_STORE SET METADATA_VALUE=? WHERE METADATA_KEY=? AND REGION=?";
+	private String replaceValueQuery = "UPDATE %sMETADATA_STORE SET METADATA_VALUE=? WHERE METADATA_KEY=? AND METADATA_VALUE=? AND REGION=?";
 
-	private String removeValueQuery = "DELETE FROM %SMETADATA_STORE WHERE METADATA_KEY=? AND REGION=?";
+	private String replaceValueByKeyQuery = "UPDATE %sMETADATA_STORE SET METADATA_VALUE=? WHERE METADATA_KEY=? AND REGION=?";
 
-	private String putIfAbsentValueQuery = "INSERT INTO %SMETADATA_STORE(METADATA_KEY, METADATA_VALUE, REGION) "
-			+ "SELECT ?, ?, ? FROM %SMETADATA_STORE WHERE METADATA_KEY=? AND REGION=? HAVING COUNT(*)=0";
+	private String removeValueQuery = "DELETE FROM %sMETADATA_STORE WHERE METADATA_KEY=? AND REGION=?";
 
-	@Override
-	public void afterPropertiesSet() throws Exception {
-		this.getValueQuery = String.format(this.getValueQuery, this.tablePrefix);
-		this.getValueForUpdateQuery = String.format(this.getValueForUpdateQuery, this.tablePrefix);
-		this.replaceValueQuery = String.format(this.replaceValueQuery, this.tablePrefix);
-		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);
-	}
+	private String putIfAbsentValueQuery = "INSERT INTO %sMETADATA_STORE(METADATA_KEY, METADATA_VALUE, REGION) "
+			+ "SELECT ?, ?, ? FROM %sMETADATA_STORE WHERE METADATA_KEY=? AND REGION=? HAVING COUNT(*)=0";
 
 	/**
 	 * Instantiate a {@link JdbcMetadataStore} using provided dataSource {@link DataSource}.
@@ -89,7 +81,7 @@ public JdbcMetadataStore(DataSource dataSource) {
 	 * @param jdbcOperations a {@link JdbcOperations}
 	 */
 	public JdbcMetadataStore(JdbcOperations jdbcOperations) {
-		Assert.notNull(jdbcOperations, "'jdbcOperations' must not be null");
+		Assert.notNull(jdbcOperations, "'jdbcOperations' must not be null.");
 		this.jdbcTemplate = jdbcOperations;
 	}
 
@@ -100,7 +92,7 @@ public JdbcMetadataStore(JdbcOperations jdbcOperations) {
 	 * @param tablePrefix the tablePrefix to set
 	 */
 	public void setTablePrefix(String tablePrefix) {
-		Assert.notNull(tablePrefix, "'tablePrefix' must not be null");
+		Assert.notNull(tablePrefix, "'tablePrefix' must not be null.");
 		this.tablePrefix = tablePrefix;
 	}
 
@@ -112,10 +104,33 @@ public void setTablePrefix(String tablePrefix) {
 	 * @param region the region name to set
 	 */
 	public void setRegion(String region) {
-		Assert.hasText(region, "Region must not be null or empty.");
+		Assert.hasText(region, "'region' must not be null or empty.");
 		this.region = 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)}.
+	 * @param lockHint the RDBMS vendor-specific lock hint.
+	 * @since 5.0.7
+	 */
+	public void setLockHint(String lockHint) {
+		Assert.notNull(lockHint, "'lockHint' cannot be null.");
+		this.lockHint = lockHint;
+	}
+
+	@Override
+	public void afterPropertiesSet() {
+		this.getValueQuery = String.format(this.getValueQuery, this.tablePrefix);
+		this.getValueForUpdateQuery = String.format(this.getValueForUpdateQuery, this.tablePrefix, this.lockHint);
+		this.replaceValueQuery = String.format(this.replaceValueQuery, this.tablePrefix);
+		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);
+	}
+
 	@Override
 	@Transactional
 	public String putIfAbsent(String key, String value) {

src/reference/asciidoc/jdbc.adoc

@@ -320,6 +320,7 @@ The following example show to use the `query` attribute:
     reply-channel="output"
     data-source="dataSource"/>
 ----
+====
 
 [IMPORTANT]
 ====
@@ -328,10 +329,6 @@ You can adjust this behavior with the `max-rows` option.
 If you need to return all the rows from the SELECT, consider specifying `max-rows="0"`.
 ====
 
-IMPORTANT: By default, the component for the `SELECT` query returns only the first row from the cursor.
-You can adjust this by setting the `max-rows-per-poll` option.
-If you need to return all the rows from the `SELECT`, you can set `max-rows-per-poll="0"`.
-
 As with the channel adapters, you can also provide `SqlParameterSourceFactory` instances for request and reply.
 The default is the same as for the outbound adapter, so the request message is available as the root of an expression.
 If `keys-generated="true"`, the root of the expression is the generated keys (a map if there is only one or a list of maps if multi-valued).
@@ -1111,3 +1108,7 @@ Transaction management must use `JdbcMetadataStore`.
 Inbound channel adapters can be supplied with a reference to the `TransactionManager` in the poller configuration.
 Unlike non-transactional `MetadataStore` implementations, with `JdbcMetadataStore`, the entry appears in the target table only after the transaction commits.
 When a rollback occurs, no entries are added to the `INT_METADATA_STORE` table.
+
+Since version 5.0.7, the `JdbcMetadataStore` can be configured with the RDBMS vendor-specific `lockHint` option for lock-based queries on metadata store entries.
+It is `FOR UPDATE` by default and can be configured with an empty string, if the target data base doesn't support row locking functionality.
+Please, consult with your vendor for particular possible hint in the `SELECT` expression for locking rows before updates.