DATACASS-253 - Add synchronization to CachedPreparedStatementCreator.

pom.xml

@@ -14,7 +14,7 @@
 
 <groupId>org.springframework.data</groupId>
 <artifactId>spring-data-cassandra-parent</artifactId>
-<version>1.5.0.BUILD-SNAPSHOT</version>
+<version>1.5.0.DATACASS-253-SNAPSHOT</version>
 <packaging>pom</packaging>
 
 <name>Spring Data Cassandra</name>
@@ -76,6 +76,7 @@
 <dist.id>spring-data-cassandra</dist.id>
 <el.version>1.0</el.version>
 <failsafe.version>2.16</failsafe.version>
+<multithreadedtc.version>1.01</multithreadedtc.version>
 <project.type>multi</project.type>
 <springdata.commons>1.13.0.BUILD-SNAPSHOT</springdata.commons>
 </properties>
@@ -189,6 +190,13 @@
 <version>3.1</version>
 <scope>test</scope>
 </dependency>
+
+<dependency>
+<groupId>edu.umd.cs.mtc</groupId>
+<artifactId>multithreadedtc</artifactId>
+<version>${multithreadedtc.version}</version>
+<scope>test</scope>
+</dependency>
 </dependencies>
 </dependencyManagement>
 

spring-cql/pom.xml

@@ -8,7 +8,7 @@
 <parent>
 <groupId>org.springframework.data</groupId>
 <artifactId>spring-data-cassandra-parent</artifactId>
-<version>1.5.0.BUILD-SNAPSHOT</version>
+<version>1.5.0.DATACASS-253-SNAPSHOT</version>
 <relativePath>../pom.xml</relativePath>
 </parent>
 
@@ -87,6 +87,11 @@
 <artifactId>commons-lang3</artifactId>
 <scope>test</scope>
 </dependency>
+<dependency>
+<groupId>edu.umd.cs.mtc</groupId>
+<artifactId>multithreadedtc</artifactId>
+<scope>test</scope>
+</dependency>
 </dependencies>
 
 <build>

spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2014 the original author or authors.
+ * Copyright 2013-2016 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.
@@ -27,57 +27,100 @@
 import com.datastax.driver.core.exceptions.DriverException;
 
 /**
- * This Prepared Statement Creator maintains a cache of all prepared statements for the duration of this life of the
- * container. When preparing statements with Cassandra, each Statement should be prepared once and only once due to the
- * overhead of preparing the statement.
+ * This {@link PreparedStatementCreator} maintains a static cache of all prepared statements for the duration of the JVM
+ * runtime, more specific the lifecycle of the associated {@link ClassLoader}. When preparing statements with Cassandra,
+ * each Statement should be prepared once and only once due to the overhead of preparing the statement.
+ * <p>
+ * {@link CachedPreparedStatementCreator} is thread-safe and does not require external synchronization when used by
+ * concurrent threads.
  * 
  * @author David Webb
+ * @author Mark Paluch
  */
 public class CachedPreparedStatementCreator implements PreparedStatementCreator {
 
 private static final Logger log = LoggerFactory.getLogger(CachedPreparedStatementCreator.class);
+private static final Map<Session, Map<String, PreparedStatement>> CACHE = new ConcurrentHashMap<Session, Map<String, PreparedStatement>>();
 
 private final String cql;
 
-private static final Map<Session, Map<String, PreparedStatement>> psMap = new ConcurrentHashMap<Session, Map<String, PreparedStatement>>();
-
 /**
- * Create a PreparedStatementCreator from the provided CQL.
+ * Create a {@link PreparedStatementCreator} from the provided CQL.
  * 
- * @param cql
+ * @param cql must not be empty and not {@literal null}.
  */
 public CachedPreparedStatementCreator(String cql) {
-Assert.notNull(cql, "CQL is required to create a PreparedStatement");
+
+Assert.hasText(cql, "CQL is required to create a PreparedStatement");
+
 this.cql = cql;
 }
 
 public String getCql() {
 return this.cql;
 }
 
+/* (non-Javadoc)
+ * @see org.springframework.cassandra.core.PreparedStatementCreator#createPreparedStatement(com.datastax.driver.core.Session)
+ */
 @Override
 public PreparedStatement createPreparedStatement(Session session) throws DriverException {
 
-StringBuilder keyspaceCQLKey = new StringBuilder().append(session.getLoggedKeyspace()).append("|").append(this.cql);
+StringBuilder cacheKey = new StringBuilder().append(session.getLoggedKeyspace()).append("|").append(this.cql);
+
+log.debug("Cachable PreparedStatement in Keyspace {}", session.getLoggedKeyspace());
+
+Map<String, PreparedStatement> sessionCache = getOrCreateSessionLocalCache(session);
+return getOrPrepareStatement(session, cacheKey.toString(), sessionCache);
+}
+
+private Map<String, PreparedStatement> getOrCreateSessionLocalCache(Session session) {
 
-log.debug(String.format("Cachable PreparedStatement in Keyspace [%s]", session.getLoggedKeyspace()));
+Map<String, PreparedStatement> sessionMap = CACHE.get(session);
 
-Map<String, PreparedStatement> sessionMap = psMap.get(session);
 if (sessionMap == null) {
-sessionMap = new ConcurrentHashMap<String, PreparedStatement>();
-psMap.put(session, sessionMap);
+
+synchronized (session) {
+
+if (CACHE.containsKey(session)) {
+sessionMap = CACHE.get(session);
+} else {
+
+sessionMap = new ConcurrentHashMap<String, PreparedStatement>();
+CACHE.put(session, sessionMap);
+}
+}
 }
 
-PreparedStatement pstmt = sessionMap.get(keyspaceCQLKey.toString());
+return sessionMap;
+}
+
+private PreparedStatement getOrPrepareStatement(Session session, String cacheKey,
+Map<String, PreparedStatement> sessionCache) {
+
+PreparedStatement pstmt = sessionCache.get(cacheKey);
+
 if (pstmt == null) {
-log.debug("No Cached PreparedStatement found...Creating and Caching");
-pstmt = session.prepare(this.cql);
-sessionMap.put(keyspaceCQLKey.toString(), pstmt);
-} else {
-log.debug("Found cached PreparedStatement");
+
+synchronized (sessionCache) {
+
+if (sessionCache.containsKey(cacheKey)) {
+
+log.debug("Found cached PreparedStatement");
+return sessionCache.get(cacheKey);
+}
+
+log.debug("No Cached PreparedStatement found...Creating and Caching");
+
+pstmt = session.prepare(this.cql);
+sessionCache.put(cacheKey, pstmt);
+
+return pstmt;
+}
 }
 
+log.debug("Found cached PreparedStatement");
+
 return pstmt;
 }
-
 }

spring-cql/src/test/java/org/springframework/cassandra/core/CachedPreparedStatementCreatorUnitTests.java

@@ -0,0 +1,187 @@
+/*
+ * Copyright 2016 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cassandra.core;
+
+import static edu.umd.cs.mtc.TestFramework.*;
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.*;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+
+import edu.umd.cs.mtc.MultithreadedTestCase;
+
+/**
+ * Unit tests for {@link CachedPreparedStatementCreator}.
+ * 
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class CachedPreparedStatementCreatorUnitTests {
+
+PreparedStatement preparedStatement;
+@Mock Session sessionMock;
+
+@Before
+public void before() throws Exception {
+
+preparedStatement = newProxy(PreparedStatement.class, new TestInvocationHandler());
+when(sessionMock.prepare(anyString())).thenReturn(preparedStatement);
+}
+
+/**
+ * @see DATACASS-253
+ */
+@Test(expected = IllegalArgumentException.class)
+public void shouldRejectEmptyCql() {
+new CachedPreparedStatementCreator("");
+}
+
+/**
+ * @see DATACASS-253
+ */
+@Test(expected = IllegalArgumentException.class)
+public void shouldRejectNullCql() {
+new CachedPreparedStatementCreator(null);
+}
+
+/**
+ * @see DATACASS-253
+ */
+@Test
+public void shouldCreatePreparedStatement() {
+
+CachedPreparedStatementCreator cachedPreparedStatementCreator = new CachedPreparedStatementCreator("my cql");
+
+PreparedStatement result = cachedPreparedStatementCreator.createPreparedStatement(sessionMock);
+
+assertThat(result, is(sameInstance(preparedStatement)));
+verify(sessionMock).prepare("my cql");
+}
+
+/**
+ * @see DATACASS-253
+ */
+@Test
+public void shouldCacheCreatePreparedStatement() {
+
+CachedPreparedStatementCreator cachedPreparedStatementCreator = new CachedPreparedStatementCreator("my cql");
+
+cachedPreparedStatementCreator.createPreparedStatement(sessionMock);
+cachedPreparedStatementCreator.createPreparedStatement(sessionMock);
+PreparedStatement result = cachedPreparedStatementCreator.createPreparedStatement(sessionMock);
+
+assertThat(result, is(sameInstance(preparedStatement)));
+verify(sessionMock, times(1)).prepare("my cql");
+}
+
+/**
+ * @see DATACASS-253
+ * @throws Throwable
+ */
+@Test
+public void concurrentAccessToCreateStatementShouldBeSynchronized() throws Throwable {
+
+CreatePreparedStatementIsThreadSafe concurrentPrepareStatement = new CreatePreparedStatementIsThreadSafe(
+preparedStatement, new CachedPreparedStatementCreator("my cql"));
+
+runManyTimes(concurrentPrepareStatement, 5);
+}
+
+@SuppressWarnings("unused")
+private static class CreatePreparedStatementIsThreadSafe extends MultithreadedTestCase {
+
+final CachedPreparedStatementCreator preparedStatementCreator;
+final Session session;
+final AtomicInteger atomicInteger = new AtomicInteger();
+
+public CreatePreparedStatementIsThreadSafe(final PreparedStatement preparedStatement,
+CachedPreparedStatementCreator preparedStatementCreator) {
+
+this.preparedStatementCreator = preparedStatementCreator;
+
+this.session = newProxy(Session.class, new TestInvocationHandler() {
+
+@Override
+public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+
+if (method.getName().equals("prepare") && args.length == 1) {
+
+waitForTick(2);
+atomicInteger.incrementAndGet();
+return preparedStatement;
+}
+
+return super.invoke(proxy, method, args);
+}
+});
+}
+
+public void thread1() {
+
+waitForTick(1);
+
+preparedStatementCreator.createPreparedStatement(session);
+
+assertThat(atomicInteger.get(), is(1));
+
+}
+
+public void thread2() {
+
+waitForTick(1);
+
+preparedStatementCreator.createPreparedStatement(session);
+
+assertThat(atomicInteger.get(), is(1));
+}
+}
+
+private static class TestInvocationHandler implements InvocationHandler {
+
+@Override
+public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+
+if (method.getName().equals("hashCode")) {
+return hashCode();
+}
+
+if (method.getName().equals("equals") && args.length == 1) {
+return equals(args[0]);
+}
+
+return null;
+}
+}
+
+private static <T> T newProxy(Class<T> theClass, InvocationHandler invocationHandler) {
+return (T) Proxy.newProxyInstance(CachedPreparedStatementCreatorUnitTests.class.getClassLoader(),
+new Class[] { theClass }, invocationHandler);
+}
+}

spring-data-cassandra-distribution/pom.xml

@@ -13,7 +13,7 @@
 <parent>
 <groupId>org.springframework.data</groupId>
 <artifactId>spring-data-cassandra-parent</artifactId>
-<version>1.5.0.BUILD-SNAPSHOT</version>
+<version>1.5.0.DATACASS-253-SNAPSHOT</version>
 <relativePath>../pom.xml</relativePath>
 </parent>
 

spring-data-cassandra/pom.xml

@@ -8,7 +8,7 @@
 <parent>
 <groupId>org.springframework.data</groupId>
 <artifactId>spring-data-cassandra-parent</artifactId>
-<version>1.5.0.BUILD-SNAPSHOT</version>
+<version>1.5.0.DATACASS-253-SNAPSHOT</version>
 <relativePath>../pom.xml</relativePath>
 </parent>