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

INT-4308: Document JdbcMetadataStore

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4308

* Document `JdbcMetadataStore`
* Remove unnecessary context DERBY config for the `PersistentAcceptOnceFileListFilterExternalStoreTests`
* Rework the `testFileSystemWithJdbcMetadataStore()` to use `EmbeddedDatabase` directly, without context
This commit is contained in:
Artem Bilan
2017-06-29 15:23:28 -04:00
committed by Gary Russell
parent f52fdd85ab
commit e475d9c695
9 changed files with 122 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
build.gradle
View File

@@ -344,8 +344,7 @@ project('spring-integration-file') {
testCompile project(":spring-integration-redis").sourceSets.test.output
testCompile project(":spring-integration-gemfire")
testCompile project(":spring-integration-jdbc")
testCompile "org.apache.derby:derby:$derbyVersion"
testCompile "org.apache.derby:derbyclient:$derbyVersion"
testCompile "com.h2database:h2:$h2Version"
testCompile "redis.clients:jedis:$jedisVersion"
testCompile "io.projectreactor:reactor-test:$reactorVersion"
}

4
spring-integration-file/src/test/java/int-derby.properties
View File

@@ -1,4 +0,0 @@
# Placeholders for Derby:
int.drop.script=classpath:/org/springframework/integration/jdbc/schema-drop-derby.sql
int.schema.script=classpath:/org/springframework/integration/jdbc/schema-derby.sql
int.database.incrementer.class=org.springframework.jdbc.support.incrementer.DerbyMaxValueIncrementer

26
spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests-context.xml
View File

@@ -1,26 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<jdbc:embedded-database id="dataSource" type="DERBY"/>
<jdbc:initialize-database data-source="dataSource" ignore-failures="DROPS">
<jdbc:script location="${int.drop.script}"/>
<jdbc:script location="${int.schema.script}"/>
</jdbc:initialize-database>
<context:property-placeholder location="int-${ENVIRONMENT:derby}.properties"
system-properties-mode="OVERRIDE"
ignore-unresolvable="true"
order="1"/>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>
</beans>

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

@@ -20,19 +20,18 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.sql.DataSource;
import org.apache.geode.cache.CacheFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.integration.gemfire.metadata.GemfireMetadataStore;
@@ -41,24 +40,21 @@ import org.springframework.integration.metadata.ConcurrentMetadataStore;
import org.springframework.integration.redis.metadata.RedisMetadataStore;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
/**
* @author Gary Russell
* @author Artem Bilan
* @author Bojan Vukasovic
*
* @since 4.0
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext // close at the end after class
public class PersistentAcceptOnceFileListFilterExternalStoreTests extends RedisAvailableTests {
@Autowired
private DataSource dataSource;
@Test
@RedisAvailable
public void testFileSystemWithRedisMetadataStore() throws Exception {
@@ -84,9 +80,27 @@ public class PersistentAcceptOnceFileListFilterExternalStoreTests extends RedisA
@Test
public void testFileSystemWithJdbcMetadataStore() throws Exception {
EmbeddedDatabase dataSource = new EmbeddedDatabaseBuilder()
.setType(EmbeddedDatabaseType.H2)
.addScript("classpath:/org/springframework/integration/jdbc/schema-drop-h2.sql")
.addScript("classpath:/org/springframework/integration/jdbc/schema-h2.sql")
.build();
JdbcMetadataStore metadataStore = new JdbcMetadataStore(dataSource);
metadataStore.afterPropertiesSet();
this.testFileSystem(metadataStore);
try {
testFileSystem(metadataStore);
List<Map<String, Object>> metaData = new JdbcTemplate(dataSource)
.queryForList("SELECT * FROM INT_METADATA_STORE");
assertEquals(1, metaData.size());
assertEquals("43", metaData.get(0).get("METADATA_VALUE"));
}
finally {
dataSource.shutdown();
}
}
private void testFileSystem(ConcurrentMetadataStore store) throws Exception {

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

@@ -28,11 +28,16 @@ import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
/**
* Implementation of {@link MetadataStore} using a relational database via JDBC. SQL scripts to create the necessary
* tables are packaged as <code>org/springframework/integration/jdbc/schema-*.sql</code>, where <code>*</code> is the
* target database type.
* Implementation of {@link MetadataStore} using a relational database via JDBC.
* SQL scripts to create the necessary tables are packaged as
* <code>org/springframework/integration/jdbc/schema-*.sql</code>,
* where <code>*</code> is the target database type.
* <p>
* The transaction management is required to use this {@link MetadataStore}.
*
* @author Bojan Vukasovic
* @author Artem Bilan
*
* @since 5.0
*/
public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingBean {
@@ -42,6 +47,8 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
*/
public static final String DEFAULT_TABLE_PREFIX = "INT_";
private final JdbcOperations jdbcTemplate;
private volatile String tablePrefix = DEFAULT_TABLE_PREFIX;
private volatile String region = "DEFAULT";
@@ -59,8 +66,6 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
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 final JdbcOperations jdbcTemplate;
@Override
public void afterPropertiesSet() throws Exception {
this.getValueQuery = String.format(this.getValueQuery, this.tablePrefix);
@@ -89,19 +94,21 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
}
/**
* Public setter for the table prefix property. This will be prefixed to all the table names before queries are
* Public setter for the table prefix property.
* This will be prefixed to all the table names before queries are
* executed. Defaults to {@link #DEFAULT_TABLE_PREFIX}.
*
* @param tablePrefix the tablePrefix to set
*/
public void setTablePrefix(String tablePrefix) {
Assert.notNull(tablePrefix, "'tablePrefix' must not be null");
this.tablePrefix = tablePrefix;
}
/**
* A unique grouping identifier for all messages persisted with this store. Using multiple regions allows the store
* to be partitioned (if necessary) for different purposes. Defaults to <code>DEFAULT</code>.
*
* A unique grouping identifier for all messages persisted with this store.
* Using multiple regions allows the store
* to be partitioned (if necessary) for different purposes.
* Defaults to <code>DEFAULT</code>.
* @param region the region name to set
*/
public void setRegion(String region) {
@@ -134,7 +141,8 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
}
private int tryToPutIfAbsent(String key, String value) {
return this.jdbcTemplate.update(this.putIfAbsentValueQuery, ps -> {
return this.jdbcTemplate.update(this.putIfAbsentValueQuery,
ps -> {
ps.setString(1, key);
ps.setString(2, value);
ps.setString(3, this.region);
@@ -149,12 +157,13 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(oldValue, "'oldValue' cannot be null");
Assert.notNull(newValue, "'newValue' cannot be null");
int affectedRows = this.jdbcTemplate.update(this.replaceValueQuery, ps -> {
ps.setString(1, newValue);
ps.setString(2, key);
ps.setString(3, oldValue);
ps.setString(4, this.region);
});
int affectedRows = this.jdbcTemplate.update(this.replaceValueQuery,
ps -> {
ps.setString(1, newValue);
ps.setString(2, key);
ps.setString(3, oldValue);
ps.setString(4, this.region);
});
return affectedRows > 0;
}
@@ -174,14 +183,15 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
}
catch (EmptyResultDataAccessException e) {
//if there are no rows with this key, somebody deleted it in between two calls
continue; //try to insert again from beginning
continue; //try to insert again from beginning
}
//lock successful, so - replace
this.jdbcTemplate.update(this.replaceValueByKeyQuery, ps -> {
ps.setString(1, value);
ps.setString(2, key);
ps.setString(3, this.region);
});
this.jdbcTemplate.update(this.replaceValueByKeyQuery,
ps -> {
ps.setString(1, value);
ps.setString(2, key);
ps.setString(3, this.region);
});
}
return;
}
@@ -220,4 +230,5 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
}
return null;
}
}

1
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStoreTests.java
View File

@@ -105,4 +105,5 @@ public class JdbcMetadataStoreTests {
String bar = metadataStore.get("foo");
assertEquals("bar", bar);
}
}

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

@@ -988,3 +988,52 @@ Therefore `prefix` property must be used on the `DefaultLockRepository` bean def
Sometimes it happens that one application has moved to the state when it can't release distributed lock - remove the particular record in the data base.
For this purpose such dead locks can be expired by the other application on the next locking invocation.
The `timeToLive` (TTL) option on the `DefaultLockRepository` is provided for this purpose.
[[jdbc-metadata-store]]
=== JDBC Metadata Store
Starting with _version 5.0_, the JDBC `MetadataStore` (<<metadata-store>>) implementation is available.
The `JdbcMetadataStore` can be used to maintain metadata state across application restarts.
This `MetadataStore` implementation can be used with adapters such as:
* <<twitter-inbound>>
* <<feed-inbound-channel-adapter>>
* <<file-reading>>
* <<ftp-inbound>>
* <<sftp-inbound>>
In order to configure these adapters to use the `JdbcMetadataStore`, simply declare a Spring bean using the
bean name *metadataStore*. The _Twitter Inbound Channel Adapter_ and the _Feed Inbound Channel Adapter_ will both
automatically pick up and use the declared `JdbcMetadataStore`:
[source,java]
----
@Bean
public MetadataStore metadataStore(DataSource dataSource) {
return new JdbcMetadataStore(dataSource);
}
----
Data base schema scripts for several RDMBS vendors are located in the `org.springframework.integration.jdbc` package.
For example the H2 DDL for metadata table looks like:
[source,sql]
----
CREATE TABLE INT_METADATA_STORE (
METADATA_KEY VARCHAR(255) NOT NULL,
METADATA_VALUE VARCHAR(4000),
REGION VARCHAR(100) NOT NULL,
constraint METADATA_STORE primary key (METADATA_KEY, REGION)
);
----
The `INT_` prefix can be changed according to the target data base design requirements and the `JdbcMetadataStore` can be configured to use the custom prefix.
The `JdbcMetadataStore` implements `ConcurrentMetadataStore`, allowing it to be reliably shared across multiple
application instances where only one instance will be allowed to store or modify a key's value.
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.

3
src/reference/asciidoc/meta-data-store.adoc
View File

@@ -14,9 +14,10 @@ If you need to persist metadata between Application Context restarts, these pers
the framework:
* `PropertiesPersistingMetadataStore`
* <<redis-metadata-store>>
* <<gemfire-metadata-store>>
* <<jdbc-metadata-store>>
* <<mongodb-metadata-store>>
* <<redis-metadata-store>>
* <<zk-metadata-store>>

7
src/reference/asciidoc/whats-new.adoc
View File

@@ -47,6 +47,13 @@ The `ErrorMessagePublisher` and the `ErrorMessageStrategy` are provided for crea
See <<namespace-errorhandler>> for more information.
==== JDBC Metadata Store
A JDBC implementation of `MetadataStore` implementation is now provided.
This is useful when it is necessary to ensure transactional boundaries for metadata.
See <<jdbc-metadata-store>> for more information.
[[x5.0-general]]
=== General Changes