Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4499: Add JdbcMetadataStore.setLockHint()

Browse Source 
JIRA https://jira.spring.io/browse/INT-4499
Fixes https://github.com/spring-projects/spring-integration/issues/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**
This commit is contained in:
Artem Bilan
2018-07-11 13:51:24 -04:00
committed by Gary Russell
parent bf1b600965
commit 2126e9b9b5
3 changed files with 47 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 class PersistentAcceptOnceFileListFilterExternalStoreTests extends RedisA
.build();
JdbcMetadataStore metadataStore = new JdbcMetadataStore(dataSource);
metadataStore.setLockHint("");
metadataStore.afterPropertiesSet();
try {

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

@@ -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 class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
* @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 class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
* @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 class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
* @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) {

9
src/reference/asciidoc/jdbc.adoc
View File

@@ -298,6 +298,7 @@ The reply message is then generated from the result, like the inbound adapter, a
reply-channel="output"
data-source="dataSource"/>
----
====
[IMPORTANT]
====
@@ -306,7 +307,7 @@ This can be adjusted with the `max-rows-per-poll` option.
Consider to specify `max-rows-per-poll="0"` if you need to return all the rows from the SELECT.
====
As with the channel adapters, there is also the option to provide `SqlParameterSourceFactory` instances for request and reply.
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"` then the root of the expression is the generated keys (a map if there is only one or a list of maps if multi-valued).
@@ -1095,4 +1096,8 @@ All of these operations are _atomic_ via transaction guarantees.
Transaction management is required to 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 is added to the `INT_METADATA_STORE` table.
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.