diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java index 3c2ddc1194..2b37cf07ee 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java +++ b/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 template = new RedisTemplate(); + RedisTemplate 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 { diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStore.java index 34414a5081..dac76bb583 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStore.java +++ b/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 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) { diff --git a/src/reference/asciidoc/jdbc.adoc b/src/reference/asciidoc/jdbc.adoc index be604f284a..f9b54a13e4 100644 --- a/src/reference/asciidoc/jdbc.adoc +++ b/src/reference/asciidoc/jdbc.adoc @@ -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.