The use of CacheLoader or CacheStore requires the backing map for the partitioned cache to be configured as a read-write backing map. A read-write backing map provides many configuration options for CacheLoader and CacheStore that will soon be covered in detail.

The following is a sample configuration for the CacheLoader:

<distributed-scheme>
<scheme-name>partitioned</scheme-name>
<backing-map-scheme>
<read-write-backing-map-scheme>
<internal-cache-scheme>
<local-scheme />
</internal-cache-scheme>
<cachestore-scheme>
<class-scheme>
<class-name>
com.seovic.coherence.book.ch8.AccountCacheLoader
</class-name>
</class-scheme>
</cachestore-scheme>
</read-write-backing-map-scheme>
</backing-map-scheme>
<autostart>true</autostart>
</distributed-scheme>

Some points to note:

Although this CacheLoader will work perfectly well in development, it has several flaws that make it unsuitable for use in production, which are as follows:

Due to these issues, it may be desirable (but not required) to use a container to manage the configuration and shared infrastructure required for database connectivity. This can be accomplished with a standard JEE container such as Oracle WebLogic Server or Apache Tomcat.

It can also be accomplished with a non-JEE container such as the Spring Framework. For the remainder of this chapter, we will use Spring to manage cache stores and related artifacts.

Spring is a popular dependency injection framework that provides utility classes that make it easy to write JDBC code and manage a DataSource. A full tutorial on Spring is beyond the scope of this chapter; if you are unfamiliar with Spring please refer to the user guide at http://www.springsource.org/documentation.

The Coherence Tools project provides several useful classes that make both the implementation and configuration of CacheLoader and CacheStore classes much more convenient.

The first one is ConfigurableCacheStore, which allows you to configure cache stores within a Spring application context. The main benefit is that you can have a single backing map configuration for all persistent caches in your application. You configure Coherence to use ConfigurableCacheStore, which in turn delegates to the concrete CacheStore implementation defined within the Spring application context.

For example, we can add the following backing map definition to our cache configuration file and reuse it for all the caches that require persistence, as shown next:

<read-write-backing-map-scheme>
<scheme-name>persistent-backing-map</scheme-name>
<internal-cache-scheme>
<local-scheme>
<scheme-ref>unlimited-backing-map</scheme-ref>
</local-scheme>
</internal-cache-scheme>
<cachestore-scheme>
<class-scheme>
<class-name>
c.s.coherence.util.persistence.ConfigurableCacheStore
</class-name>
<init-params>
<init-param>
<param-type>string</param-type>
<param-value>{cache-name}</param-value>
</init-param>
</init-params>
</class-scheme>
</cachestore-scheme>
<write-delay>
{write-delay 0}
</write-delay>
<write-batch-factor>
{write-batch-factor 0}
</write-batch-factor>
<write-requeue-threshold>
{write-requeue-threshold 0}
</write-requeue-threshold>
<refresh-ahead-factor>
{refresh-ahead-factor 0}
</refresh-ahead-factor>
</read-write-backing-map-scheme>

The ConfigurableCacheStore constructor takes a single string argument, which is used to retrieve the concrete CacheStore to delegate to from the associated Spring context. In this example we are simply passing cache name, which implies that we will need to define a Spring bean with a matching name for each concrete cache store within the persistence context.

You should also notice that we use macro arguments to configure several read-write backing map settings at the bottom. This will allow us to change the persistence behavior of individual caches without having to create separate backing map configurations. These settings will be covered in the upcoming sections.

Now that we have a backing map definition, we can use it to configure the cache scheme for accounts, as is done in the following code:

<distributed-scheme>
<scheme-name>accounts-scheme</scheme-name>
<scheme-ref>default-partitioned</scheme-ref>
<service-name>AccountsCacheService</service-name>
<backing-map-scheme>
<read-write-backing-map-scheme>
<scheme-ref>persistent-backing-map</scheme-ref>
</read-write-backing-map-scheme>
</backing-map-scheme>
<autostart>true</autostart>
</distributed-scheme>

Another important benefit of ConfigurableCacheStore is that it allows us to use all the existing Spring constructs when configuring our cache stores, including transaction managers, data sources, connection pools, and so on. This makes cache store configuration significantly easier, as you'll see shortly.

The second useful class provided by the Coherence Tools project is AbstractJdbcCacheStore. This class uses Spring's JdbcTemplate to implement all necessary data access code, which makes implementation of concrete JDBC-based cache stores trivial; all you need to do is to define the SQL statements and RowMapper that will be used for object retrieval and persistence.

The following is a snippet of AbstractJdbcCacheStore relevant to the load operation:

@Transactional
public abstract class AbstractJdbcCacheStore<T>
extends AbstractBatchingCacheStore {
public AbstractJdbcCacheStore(DataSource dataSource) {
m_jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
protected abstract String getSelectSql();
protected abstract RowMapper<T> getRowMapper();
@Transactional(readOnly = true)
public Object load(Object key) {
List<T> results = getJdbcTemplate().query(
getSelectSql(), getRowMapper(),
getPrimaryKeyComponents(key));
return results.size() == 0 ? null : results.get(0);
}
protected Object[] getPrimaryKeyComponents(Object key) {
return new Object[] {key};
}
protected SimpleJdbcTemplate getJdbcTemplate() {
return m_jdbcTemplate;
}
private final SimpleJdbcTemplate m_jdbcTemplate;
AbstractJdbcCacheStore classAbstractJdbcCacheStore classabout}

The following is an implementation of the AccountJdbcCacheStore from our sample application, which leverages features provided by AbstractJdbcCacheStore. The parts relevant to load are included, as shown in the following code:

public class AccountJdbcCacheStore
extends AbstractJdbcCacheStore<Account> {
public AccountJdbcCacheStore(DataSource dataSource) {
super(dataSource);
}
@Override
protected String getSelectSql() {
return SELECT_SQL;
}
@Override
protected RowMapper<Account> getRowMapper() {
return ROW_MAPPER;
}
private static final RowMapper<Account> ROW_MAPPER =
new RowMapper<Account>() {
public Account mapRow(ResultSet rs, int i)
throws SQLException {
return new Account(rs.getLong("id"),
rs.getString("description"),
new Money(rs.getLong("balance"),
rs.getString("currency")),
rs.getLong("last_tx"),
rs.getLong("customer_id"));
}
};
private static final String SELECT_SQL = "SELECT id, description, balance, currency, " + "last_tx, customer_id " +
"FROM accounts WHERE id = ?";
}

Now that we have both the CacheStore and necessary cache configuration, the last thing we need to do is add the CacheStore bean definition to the Spring context defined in the persistence-context.xml file, which should be located in the root of the classpath:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns=...>
<tx:annotation-driven transaction-manager="txManager"/>
<bean id="txManager"
class="o.s.j.d.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>
<!-- H2 DataSource connection pool -->
<bean id="dataSource"
class="org.h2.jdbcx.JdbcConnectionPool"
factory-method="create">
<constructor-arg value="jdbc:h2:tcp://localhost/db/ coherent-bank-db"/>
<constructor-arg value="sa"/>
<constructor-arg value=""/>
<property name="maxConnections" value="5" />
</bean>
<!-- cache store definitions -->
<bean name="accounts"
class="c.s.s.bank.persistence.AccountJdbcCacheStore">
<constructor-arg ref="dataSource"/>
</bean>
</beans>

Note that the JDBC DataSource in this example is configured using the connection pool that ships with the H2 database we used for the sample application. If your JDBC driver does not provide its own connection pool you can use a third-party pool such as Apache DBCP (http://commons.apache.org/dbcp/) or C3P0 (http://sourceforge.net/projects/c3p0/). The Oracle Universal Connection Pool (UCP) is another free option for connection pooling, which is available at http://www.oracle.com/technology/tech/java/sqlj_jdbc/UCP_dev_guide.pdf.

The configuration above should be self-explanatory—the accounts bean is injected with a data source, which is the only dependency it requires.

Many Coherence applications do not have exclusive access to the database. In several cases the database and its contents outlive several generations of applications that create and access that data. For these types of applications, Coherence caches can be configured to expire after a configurable amount of time to limit data staleness.

The expiration of heavily accessed cached entries may cause inconsistent response times, as some requests will require a read through to the database. This is especially true for applications that have aggressive expiry configured in order to minimize the amount of time for which cache entries may remain stale. However, a read-write backing map can be configured to fetch these items from the database in a background thread before they expire from the cache.

This setting is controlled by<refresh-ahead-factor>, which is a fraction of the expiry time. If an entry is accessed from the cache during the configured refresh-ahead period, it will be refreshed from the data source in a background thread.

public class ExchangeRateRestCacheLoader
extends AbstractCacheLoader {
public Object load(Object key) {
Map<String, String> params =
Collections.singletonMap("instrument", (String) key);
RestTemplate client = new RestTemplate();
String rate = client.getForObject(m_url, String.class, params);
return new BigDecimal(rate);
}
public void setUrl(String url) {
m_url = url;
}
private String m_url;
}

<cache-mapping>
<cache-name>exchange-rates</cache-name>
<scheme-name>exchange-rates-scheme</scheme-name>
<init-params>
<init-param>
<param-name>refresh-ahead-factor</param-name>
<param-value>.75</param-value>
</init-param>
</init-params>
</cache-mapping>
...
<distributed-scheme>
<scheme-name>exchange-rates-scheme</scheme-name>
<scheme-ref>default-partitioned</scheme-ref>
<service-name>ExchangeRatesCacheService</service-name>
<backing-map-scheme>
<read-write-backing-map-scheme>
<scheme-ref>persistent-backing-map</scheme-ref>
<internal-cache-scheme>
<local-scheme>
<scheme-ref>unlimited-backing-map</scheme-ref>
<expiry-delay>1m</expiry-delay>
</local-scheme>
</internal-cache-scheme>
</read-write-backing-map-scheme>
</backing-map-scheme>
<autostart>true</autostart>
</distributed-scheme>

There are a few additional items that should be configured in the read-write backing map to support read through. First, a<high-units> setting should be configured for the<local-scheme> in<internal-cache-scheme>. This will ensure that the storage-enabled cluster member does not run out of memory while storing cached data.

Additionally, a thread pool should be configured for the partitioned cache service. Normal partitioned caching does not require a thread pool. This is because a thread pool will cause the service thread to queue up a cache request and a context switch will be required for a worker thread to process the request. Since reading from memory is such a low-overhead operation, there will be less overhead if the service thread simply processes the request.

However, the inevitability of a cache request through the read-write backing map resulting in a blocking I/O call over the network makes it a requirement to configure a thread pool. The service thread is the lifeblood of the partitioned cache; it should be kept as free as possible so it can service requests coming from other cluster members.

To start with, configure the thread count to be equal to the number of connections in the connection pool. Your application may be able to handle more threads and/or less connections; be sure to load test your configuration before going to production.

In order to configure a thread pool for the AccountsCacheService we defined earlier, we need to add a thread-count configuration element to it, which is shown next:

<distributed-scheme>
<scheme-name>accounts-scheme</scheme-name>
<scheme-ref>default-partitioned</scheme-ref>
<service-name>AccountsCacheService</service-name>
<thread-count>5</thread-count>
<backing-map-scheme>
<read-write-backing-map-scheme>
<scheme-ref>persistent-backing-map</scheme-ref>
</read-write-backing-map-scheme>
</backing-map-scheme>
<autostart>true</autostart>
</distributed-scheme>

Finally, when using read through, it may be desirable to maintain a set of cache misses. This information can be used to avoid reading through to the CacheLoader if the likelihood of the value existing in the database immediately after a cache miss is low.

This cache can be configured via the<miss-cache-scheme> in the read-write backing map. This setting requires the use of<local-scheme>, which would ideally be configured with an expiry that indicates how long the read-write backing map will wait before reissuing a load request for a given key to the CacheLoader.

As previously mentioned, Coherence ships with AbstractCacheStore, which provides a default implementation of storeAll that iterates the map passed to it and invokes store on each entry. Although this implementation works, it may not be the optimal way of storing multiple entries to the database.

With a write through configuration, this method will never be invoked by Coherence (unless operation-bundling is enabled, see the previous note on loadAll). Each entry will be written to the database via the store method, even if using putAll from the client.

On the other hand, write behind does invoke storeAll when there are multiple entries in the queue that need to be stored. Therefore, a CacheStore configured to use write behind should take advantage of any batching functionality provided by the database. Many databases and other data stores have batching features that allow for a much more efficient (in terms of both latency and throughput) storeAll implementation.

The AbstractJdbcCacheStore base class being used for our CacheStore implementations extends AbstractBatchingCacheStore, and implements an abstract method storeBatch, as shown next:

public void storeBatch(Map mapBatch) {
SqlParameterSource[] batch =
SqlParameterSourceUtils.createBatch(
mapBatch.values().toArray());
getJdbcTemplate().batchUpdate(getMergeSql(), batch);
}

This implementation uses batching via Spring's SimpleJdbcTemplate.batchUpdate method, whose underlying implementation uses JDBC PreparedStatement batching, to execute the updates. Depending on the JDBC driver implementation and latency to the database, batching updates in this manner can yield an order of magnitude performance improvement over sending one update at a time, even within the same database transaction.

In turn, storeBatch is invoked by AbstractBatchingCacheStore as follows:

public void storeAll(Map mapEntries) {
int batchSize = getBatchSize();
if (batchSize == 0 || mapEntries.size() < batchSize) {
storeBatch(mapEntries);
}
else {
Map batch = new HashMap(batchSize);
while (!mapEntries.isEmpty()) {
// since entries will be removed from mapEntries,
// the iterator needs to be recreated every time
Iterator iter = mapEntries.entrySet().iterator();
while (iter.hasNext() && batch.size() < batchSize) {
// retrieving the entry will force
// deserialization
Map.Entry entry = (Map.Entry) iter.next();
batch.put(entry.getKey(), entry.getValue());
}
storeBatch(batch);
// remove the entries we've successfully stored
mapEntries.keySet().removeAll(batch.keySet());
batch.clear();
}
}
}

AbstractBatchingCacheStore gives us more control over batching granularity, by allowing us to chunk the map passed by the write-behind thread to storeAll into batches of configurable size.

Let's say that we have determined during load testing that batches of 500 objects work best for our application. Although you can use the write batch factor to roughly determine the percentage of queued entries that will be persisted at a time, there is no way to tell Coherence exactly how many objects to send in each batch. You might get 50, or you might get 10,000.

Smaller batches are usually not a problem—they simply lighter load on the system. However, writing out a large number (thousands) of items to a database in a single batch may be suboptimal for various reasons (very large transactions requiring a large transaction/rollback log in the database, memory consumption by the JDBC driver, and so on). It will also cause Coherence to consume more memory in the storage member.

As mentioned in the previous chapters, data is stored in binary format in the backing map. This is also the case for data stored in the write-behind queue. The map passed to storeAll will deserialize its contents lazily. If a large amount of data is passed in the map and it is deserialized in its entirety to perform a batch store, the JVM may experience very long GC pauses or even run out of heap. Therefore, it is a good idea to cap the number of items written to the database in a single batch.

Fortunately, the solution is simple—implement chunking within the storeAll method and persist large batches in multiple database calls. This is exactly what AbstractBatchingCacheStore does for us; all we need to do is configure the batch size for accounts and transactions within our cache store bean definitions, as shown next:

<bean name="accounts"
class="c.s.s.b.persistence.AccountJdbcCacheStore">
<constructor-arg ref="dataSource"/>
<property name="batchSize" value="500"/>
</bean>
<bean name="transactions"
class="c.s.s.b.persistence.TransactionJdbcCacheStore">
<constructor-arg ref="dataSource"/>
<property name="batchSize" value="500"/>
</bean>

Both Coherence cluster members and persistent data stores that Coherence writes data to could fail at any time. That means that we need to be prepared to deal with such failures.

It is considered a good practice to configure backing maps with size limits to prevent an OutOfMemoryError in the cache server tier. This can present interesting challenges in a write-behind scenario.

Consider the case where a client thread performs a put into a cache that causes evictions to occur to avoid exceeding high units. What will happen if the eviction policy dictates that we evict an entry from the backing map that is still in the write-behind queue (and therefore has not been flushed to the database)?

In this situation, the read-write backing map will synchronously invoke the store operation on any entries about to be evicted. The implication here is that the client thread performing the put operation will be blocked while evicted entries are flushed to the database. It is an unfortunate side effect for the client thread, as its operation will experience a higher than expected latency, but it acts as a necessary throttle to avoid losing data.

This edge condition highlights the necessity to configure a worker thread pool even for caches that are strictly performing write behind in order to prevent this flush from occurring on the service thread. It is important to keep in mind that the store operation won't always necessarily be performed by the write-behind thread.

Note that this can also occur with caches that have expiry configured. The likelihood of this occurring will decrease if there is a large difference between expiry time and write-behind time.

The CacheStore interface defines erase and eraseAll for removing entries from a backing store. These operations are always performed synchronously, even if write behind is enabled.

That said, most enterprise databases never allow the execution of DELETE statements. Instead, records no longer in use are flagged as inactive. It is therefore expected that erase and eraseAll will be implemented sparingly.

The erase implementation of AbstractCacheStore throws UnsupportedOperationException to indicate that the method is not implemented. This is also optional for store and storeAll implementations. If the read-write backing map detects that any of these methods are not implemented, it will refrain from invoking them in the future.

Applications that use read through are most likely using a database as the system of record. If a storage member is lost, the data can be retrieved from the database when needed. Therefore, this type of application may be a good candidate for a distributed cache with backups disabled.

Disabling backups will slightly improve read through performance (as a backup copy will not be made when the object is loaded from the database) and it will increase the storage capacity of the distributed cache. The downside is increased latency to retrieve lost items from the database if a storage member fails.

To disable backups, add the following to<distributed-scheme>:

<backup-count>0</backup-count>

Although disabling backups with read through is a good combination, disabling backups with write behind is not a good idea, as it can result in data loss if a storage member fails while cache entries are in the write behind queue.

However, once the queued entries are flushed, it may be desirable to remove the backup copy for the reasons mentioned previously. This can be enabled in<distributed-scheme> with the following setting:

<backup-count-after-writebehind>
0
</backup-count-after-writebehind>

The first step is to modify the Account object with the required JPA annotations, as shown next:

@Entity
@Table(name="accounts")
public class Account
implements Serializable {
public Account() {
}
...
@Id
private long id;
...
}

There are several ways of mapping a Java class to a relational database using JPA. The previous example uses field-level mapping and specifies the table and column names that deviate from the defaults. There is no @GeneratedValue annotation as JPA will not be responsible for key generation. In general, optimistic locking should not be enabled, especially when combined with write behind.

The next step is to create the META-INF/persistence.xml file that JPA requires for configuration:

<persistence xmlns="http://java.sun.com/xml/ns/persistence"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://java.sun.com/xml/ns/persistence
http://java.sun.com/xml/ns/persistence/persistence_1_0.xsd"
version="1.0">
<persistence-unit name="CoherentBankUnit"
transaction-type="RESOURCE_LOCAL">
<provider>
org.eclipse.persistence.jpa.PersistenceProvider
</provider>
<class>com.seovic.coherence.book.ch8.Account</class>
<properties>
<property name="javax.persistence.jdbc.driver"
value="oracle.jdbc.OracleDriver"/>
<property name="javax.persistence.jdbc.url"
value="jdbc:oracle:thin:@10.211.55.4:1521:XE"/>
<property name="javax.persistence.jdbc.user"
value="scott"/>
<property name="javax.persistence.jdbc.password"
value="tiger"/>
<property name="eclipselink.cache.shared.default"
value="false"/>
<property name="eclipselink.jdbc.read-connections.min"
value="1"/>
<property name="eclipselink.jdbc.read-connections.max"
value="3"/>
<property name="eclipselink.jdbc.write-connections.min"
value="2"/>
<property name="eclipselink.jdbc.write-connections.max"
value="5"/>
<property name="eclipselink.jdbc.batch-writing"
value="JDBC"/>
<property name="eclipselink.logging.level"
value="FINEST"/>
</properties>
</persistence-unit>
</persistence>

As EclipseLink is running outside of a JEE container, the only supported transaction type is RESOURCE_LOCAL. Note that JPA entities should be listed in<class> elements.

By default L2 caching is enabled; this should be disabled (via the eclipselink.cache.shared.default property) when using EclipseLink in a CacheStore, as L2 caching would be redundant.

EclipseLink ships with a connection pool suitable for JPA applications running outside of a container, the settings for which are configured via the eclipselink.jdbc.*-connections properties indicated in persistence.xml previously. To enable batching for storeAll operations, the eclipselink.jdbc.batch-writing property is set for JDBC PreparedStatement batching. For development/testing purposes, logging is set to the highest setting (to produce the most output); this should be adjusted accordingly when deploying to production.

Finally, the following modifications should be made to coherence-cache-config.xml:

<class-scheme>
<class-name>
com.tangosol.coherence.jpa.JpaCacheStore
</class-name>
<init-params>
<!-- Entity name -->
<init-param>
<param-type>java.lang.String</param-type>
<param-value>account</param-value>
</init-param>
<!-- Fully qualified entity class name -->
<init-param>
<param-type>java.lang.String</param-type>
<param-value>
com.seovic.samples.bank.domain.Account
</param-value>
</init-param>
<!-- Persistence unit name -->
<init-param>
<param-type>java.lang.String</param-type>
<param-value>CoherentBankUnit</param-value>
</init-param>
</init-params>
</class-scheme>

Note the name of the CacheStore class, com.tangosol.coherence.jpa.JpaCacheStore. This class ships with Coherence, but in order to use it, you will need to add coherence-jpa.jar to the classpath.

The JpaCacheStore constructor requires the following parameters:

As we are using ConfigurableCacheStore, we can optionally leave the Coherence cache configuration alone, and simply change the definition of our accounts bean within persistence-context.xml, as shown next:

<bean name="accounts"
class="com.tangosol.coherence.jpa.JpaCacheStore">
<constructor-arg value="Account"/>
<constructor-arg value="com.seovic.samples.bank.domain.Account"/>
<constructor-arg value="CoherentBankUnit"/>
</bean>

This simplified configuration for JpaCacheStore is a further demonstration of the benefits of using ConfigurableCacheStore in Spring-based Coherence applications.