DATACASS-330 - Review and apply additional polish.

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

@@ -172,7 +172,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 			ResultSetFuture results = getCurrentSession().executeAsync(simpleStatement);
 
 			return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
-					new GuavaListenableFutureAdapter<>(results, ex -> translateExceptionIfPossible("Query", cql, ex)),
+					new GuavaListenableFutureAdapter<>(results,
+							ex -> translateExceptionIfPossible("Query", cql, ex)),
 					resultSetExtractor::extractData), getExceptionTranslator());
 		} catch (DriverException e) {
 			throw translateException("Query", cql, e);
@@ -188,8 +189,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 
 		ListenableFuture<?> results = query(cql, newResultSetExtractor(rowCallbackHandler));
 
-		return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
-				getExceptionTranslator());
+		return new ExceptionTranslatingListenableFutureAdapter<>(
+				new MappingListenableFutureAdapter<>(results, o -> null), getExceptionTranslator());
 	}
 
 	/*
@@ -247,7 +248,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 		ListenableFuture<List<T>> results = query(cql, newResultSetExtractor(rowMapper));
 
 		return new ExceptionTranslatingListenableFutureAdapter<>(
-				new MappingListenableFutureAdapter<>(results, DataAccessUtils::requiredSingleResult), getExceptionTranslator());
+				new MappingListenableFutureAdapter<>(results, DataAccessUtils::requiredSingleResult),
+				getExceptionTranslator());
 	}
 
 	/*
@@ -314,8 +316,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 
 		ListenableFuture<?> result = query(statement, newResultSetExtractor(rowCallbackHandler));
 
-		return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(result, o -> null),
-				getExceptionTranslator());
+		return new ExceptionTranslatingListenableFutureAdapter<>(
+				new MappingListenableFutureAdapter<>(result, o -> null), getExceptionTranslator());
 	}
 
 	/*
@@ -375,7 +377,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 		ListenableFuture<List<T>> results = query(statement, newResultSetExtractor(rowMapper));
 
 		return new ExceptionTranslatingListenableFutureAdapter<>(
-				new MappingListenableFutureAdapter<>(results, DataAccessUtils::requiredSingleResult), getExceptionTranslator());
+				new MappingListenableFutureAdapter<>(results, DataAccessUtils::requiredSingleResult),
+						getExceptionTranslator());
 	}
 
 	/*
@@ -398,6 +401,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 	@Override
 	public ListenableFuture<Boolean> execute(AsyncPreparedStatementCreator preparedStatementCreator)
 			throws DataAccessException {
+
 		return query(preparedStatementCreator, ResultSet::wasApplied);
 	}
 
@@ -417,6 +421,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 	@Override
 	public ListenableFuture<Boolean> execute(String cql, PreparedStatementBinder preparedStatementBinder)
 			throws DataAccessException {
+
 		return query(newAsyncPreparedStatementCreator(cql), preparedStatementBinder, ResultSet::wasApplied);
 	}
 
@@ -446,10 +451,12 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 			}
 
 			Session currentSession = getCurrentSession();
+
 			return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
 					preparedStatementCreator.createPreparedStatement(currentSession), preparedStatement -> {
 						try {
-							return action.doInPreparedStatement(currentSession, applyStatementSettings(preparedStatement));
+							return action.doInPreparedStatement(currentSession,
+									applyStatementSettings(preparedStatement));
 						} catch (DriverException e) {
 							throw translateException("PreparedStatementCallback", preparedStatement.toString(), e);
 						}
@@ -479,10 +486,11 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 	public ListenableFuture<Void> query(AsyncPreparedStatementCreator preparedStatementCreator,
 			RowCallbackHandler rowCallbackHandler) throws DataAccessException {
 
-		ListenableFuture<?> results = query(preparedStatementCreator, null, newResultSetExtractor(rowCallbackHandler));
+		ListenableFuture<?> results = query(preparedStatementCreator, null,
+				newResultSetExtractor(rowCallbackHandler));
 
-		return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
-				getExceptionTranslator());
+		return new ExceptionTranslatingListenableFutureAdapter<>(
+				new MappingListenableFutureAdapter<>(results, o -> null), getExceptionTranslator());
 	}
 
 	/*
@@ -510,13 +518,14 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 
 		try {
 			if (logger.isDebugEnabled()) {
-				logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
+				logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator),
+						preparedStatementCreator);
 			}
 
 			Session session = getCurrentSession();
 
-			PersistenceExceptionTranslator exceptionTranslator = ex -> translateExceptionIfPossible("Query",
-					toCql(preparedStatementCreator), ex);
+			PersistenceExceptionTranslator exceptionTranslator = ex ->
+					translateExceptionIfPossible("Query", toCql(preparedStatementCreator), ex);
 
 			ListenableFuture<BoundStatement> statementFuture = new MappingListenableFutureAdapter<>(
 					preparedStatementCreator.createPreparedStatement(session), preparedStatement -> {
@@ -530,30 +539,31 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 
 			SettableListenableFuture<T> settableListenableFuture = new SettableListenableFuture<>();
 
-			statementFuture.addCallback(
-					boundStatement -> Futures.addCallback(session.executeAsync(boundStatement), new FutureCallback<ResultSet>() {
+			statementFuture.addCallback(boundStatement ->
+					Futures.addCallback(session.executeAsync(boundStatement), new FutureCallback<ResultSet>() {
 						@Override
 						public void onSuccess(ResultSet result) {
 							try {
 								settableListenableFuture.set(resultSetExtractor.extractData(result));
 							} catch (DriverException e) {
-								settableListenableFuture.setException(exceptionTranslator.translateExceptionIfPossible(e));
+								settableListenableFuture.setException(
+										exceptionTranslator.translateExceptionIfPossible(e));
 							}
 						}
 
 						@Override
 						public void onFailure(Throwable ex) {
 							if (ex instanceof DriverException) {
-								settableListenableFuture
-										.setException(exceptionTranslator.translateExceptionIfPossible((DriverException) ex));
+								settableListenableFuture.setException(
+										exceptionTranslator.translateExceptionIfPossible((DriverException) ex));
 							} else {
 								settableListenableFuture.setException(ex);
 							}
 						}
 					}), ex -> {
 						if (ex instanceof DriverException) {
-							settableListenableFuture
-									.setException(exceptionTranslator.translateExceptionIfPossible((DriverException) ex));
+							settableListenableFuture.setException(
+									exceptionTranslator.translateExceptionIfPossible((DriverException) ex));
 						} else {
 							settableListenableFuture.setException(ex);
 						}
@@ -578,8 +588,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 		ListenableFuture<?> results = query(preparedStatementCreator, preparedStatementBinder,
 				newResultSetExtractor(rowCallbackHandler));
 
-		return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
-				getExceptionTranslator());
+		return new ExceptionTranslatingListenableFutureAdapter<>(
+				new MappingListenableFutureAdapter<>(results, o -> null), getExceptionTranslator());
 	}
 
 	/*
@@ -615,8 +625,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 		ListenableFuture<?> results = query(newAsyncPreparedStatementCreator(cql), newPreparedStatementBinder(args),
 				newResultSetExtractor(rowCallbackHandler));
 
-		return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
-				getExceptionTranslator());
+		return new ExceptionTranslatingListenableFutureAdapter<>(
+				new MappingListenableFutureAdapter<>(results, o -> null), getExceptionTranslator());
 	}
 
 	/*
@@ -653,8 +663,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 		ListenableFuture<?> results = query(newAsyncPreparedStatementCreator(cql), preparedStatementBinder,
 				newResultSetExtractor(rowCallbackHandler));
 
-		return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
-				getExceptionTranslator());
+		return new ExceptionTranslatingListenableFutureAdapter<>(
+				new MappingListenableFutureAdapter<>(results, o -> null), getExceptionTranslator());
 	}
 
 	/*
@@ -807,7 +817,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 		@Override
 		public ListenableFuture<PreparedStatement> createPreparedStatement(Session session) throws DriverException {
 
-			return new GuavaListenableFutureAdapter<>(session.prepareAsync(getCql()), this.persistenceExceptionTranslator);
+			return new GuavaListenableFutureAdapter<>(session.prepareAsync(getCql()),
+					this.persistenceExceptionTranslator);
 		}
 	}
 

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

@@ -431,6 +431,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 			}
 
 			Session session = getCurrentSession();
+
 			return action.doInPreparedStatement(session,
 					applyStatementSettings(preparedStatementCreator.createPreparedStatement(session)));
 
@@ -446,6 +447,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 	@Override
 	public <T> T query(PreparedStatementCreator preparedStatementCreator, ResultSetExtractor<T> resultSetExtractor)
 			throws DataAccessException {
+
 		return query(preparedStatementCreator, null, resultSetExtractor);
 	}
 
@@ -456,6 +458,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 	@Override
 	public void query(PreparedStatementCreator preparedStatementCreator, RowCallbackHandler rowCallbackHandler)
 			throws DataAccessException {
+
 		query(preparedStatementCreator, null, newResultSetExtractor(rowCallbackHandler));
 	}
 
@@ -466,6 +469,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 	@Override
 	public <T> List<T> query(PreparedStatementCreator preparedStatementCreator, RowMapper<T> rowMapper)
 			throws DataAccessException {
+
 		return query(preparedStatementCreator, null, newResultSetExtractor(rowMapper));
 	}
 
@@ -482,7 +486,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 
 		try {
 			if (logger.isDebugEnabled()) {
-				logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
+				logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator),
+						preparedStatementCreator);
 			}
 
 			Session session = getCurrentSession();
@@ -510,8 +515,9 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 	 * @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowCallbackHandler)
 	 */
 	@Override
-	public void query(PreparedStatementCreator preparedStatementCreator, PreparedStatementBinder preparedStatementBinder,
-			RowCallbackHandler rowCallbackHandler) throws DataAccessException {
+	public void query(PreparedStatementCreator preparedStatementCreator,
+			PreparedStatementBinder preparedStatementBinder, RowCallbackHandler rowCallbackHandler)
+			throws DataAccessException {
 
 		query(preparedStatementCreator, preparedStatementBinder, newResultSetExtractor(rowCallbackHandler));
 	}
@@ -532,7 +538,9 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 	 * @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.ResultSetExtractor, java.lang.Object[])
 	 */
 	@Override
-	public <T> T query(String cql, ResultSetExtractor<T> resultSetExtractor, Object... args) throws DataAccessException {
+	public <T> T query(String cql, ResultSetExtractor<T> resultSetExtractor, Object... args)
+			throws DataAccessException {
+
 		return query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args), resultSetExtractor);
 	}
 

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

@@ -15,9 +15,6 @@
  */
 package org.springframework.cassandra.core;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
 import java.util.Map;
 import java.util.function.Function;
 
@@ -31,6 +28,9 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.dao.support.DataAccessUtils;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import com.datastax.driver.core.BoundStatement;
 import com.datastax.driver.core.ConsistencyLevel;
 import com.datastax.driver.core.PreparedStatement;
@@ -256,8 +256,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	 */
 	@Override
 	public <T> Mono<T> queryForObject(String cql, RowMapper<T> rowMapper) throws DataAccessException {
-		return query(cql, rowMapper).buffer(2).flatMap(list -> Mono.just(DataAccessUtils.requiredSingleResult(list)))
-				.next();
+		return query(cql, rowMapper).buffer(2).flatMap(list ->
+				Mono.just(DataAccessUtils.requiredSingleResult(list))).next();
 	}
 
 	/* (non-Javadoc)
@@ -377,8 +377,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	 */
 	@Override
 	public <T> Mono<T> queryForObject(Statement statement, RowMapper<T> rowMapper) throws DataAccessException {
-		return query(statement, rowMapper).buffer(2).flatMap(list -> Mono.just(DataAccessUtils.requiredSingleResult(list)))
-				.next();
+		return query(statement, rowMapper).buffer(2).flatMap(list ->
+				Mono.just(DataAccessUtils.requiredSingleResult(list))).next();
 	}
 
 	/* (non-Javadoc)
@@ -567,8 +567,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	 */
 	@Override
 	public <T> Mono<T> queryForObject(String cql, RowMapper<T> rowMapper, Object... args) throws DataAccessException {
-		return query(cql, rowMapper, args).buffer(2).flatMap(list -> Mono.just(DataAccessUtils.requiredSingleResult(list)))
-				.next();
+		return query(cql, rowMapper, args).buffer(2).flatMap(list ->
+				Mono.just(DataAccessUtils.requiredSingleResult(list))).next();
 	}
 
 	/* (non-Javadoc)
@@ -611,8 +611,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 
 		Assert.hasText(cql, "CQL must not be empty");
 
-		return query(new SimpleReactivePreparedStatementCreator(cql), newArgPreparedStatementBinder(args), Mono::just)
-				.next();
+		return query(new SimpleReactivePreparedStatementCreator(cql),
+				newArgPreparedStatementBinder(args), Mono::just).next();
 	}
 
 	/* (non-Javadoc)
@@ -637,8 +637,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	 */
 	@Override
 	public Mono<Boolean> execute(String cql, PreparedStatementBinder psb) throws DataAccessException {
-		return query(new SimpleReactivePreparedStatementCreator(cql), psb, resultSet -> Mono.just(resultSet.wasApplied()))
-				.next();
+		return query(new SimpleReactivePreparedStatementCreator(cql), psb, resultSet ->
+				Mono.just(resultSet.wasApplied())).next();
 	}
 
 	/* (non-Javadoc)
@@ -666,6 +666,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 			}
 
 			BoundStatement boundStatement = newArgPreparedStatementBinder(objects).bindValues(ps);
+
 			applyStatementSettings(boundStatement);
 
 			return session.execute(boundStatement);
@@ -747,8 +748,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	 */
 	protected <T> Function<Throwable, Mono<? extends T>> translateException(String task, String cql) {
 
-		return throwable -> Mono
-				.error(throwable instanceof DriverException ? translate(task, cql, (DriverException) throwable) : throwable);
+		return throwable -> Mono.error(
+				throwable instanceof DriverException ? translate(task, cql, (DriverException) throwable) : throwable);
 	}
 
 	/**

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

@@ -15,16 +15,6 @@
  */
 package org.springframework.cassandra.core.session;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Scheduler;
-import reactor.core.scheduler.Schedulers;
-
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Scheduler;
-import reactor.core.scheduler.Schedulers;
-
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -33,7 +23,22 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.Assert;
 
-import com.datastax.driver.core.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ColumnDefinitions;
+import com.datastax.driver.core.ExecutionInfo;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.RegularStatement;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SimpleStatement;
+import com.datastax.driver.core.Statement;
 import com.google.common.util.concurrent.ListenableFuture;
 
 /**

spring-cql/src/main/java/org/springframework/cassandra/core/session/lookup/AbstractRoutingSessionFactory.java

@@ -108,7 +108,8 @@ public abstract class AbstractRoutingSessionFactory implements SessionFactory, I
 	 *          {@literal null}.
 	 */
 	public void setSessionFactoryLookup(SessionFactoryLookup sessionFactoryLookup) {
-		this.sessionFactoryLookup = (sessionFactoryLookup != null ? sessionFactoryLookup : new MapSessionFactoryLookup());
+		this.sessionFactoryLookup = (sessionFactoryLookup != null ? sessionFactoryLookup
+				: new MapSessionFactoryLookup());
 	}
 
 	/* (non-Javadoc)
@@ -129,6 +130,7 @@ public abstract class AbstractRoutingSessionFactory implements SessionFactory, I
 		Assert.notNull(this.targetSessionFactories, "Property targetSessionFactories is required");
 
 		this.resolvedSessionFactories = new HashMap<>(this.targetSessionFactories.size());
+
 		for (Map.Entry<Object, Object> entry : this.targetSessionFactories.entrySet()) {
 
 			Object lookupKey = resolveSpecifiedLookupKey(entry.getKey());
@@ -173,9 +175,9 @@ public abstract class AbstractRoutingSessionFactory implements SessionFactory, I
 		} else if (sessionFactory instanceof String) {
 			return this.sessionFactoryLookup.getSessionFactory((String) sessionFactory);
 		} else {
-			throw new IllegalArgumentException(String
-					.format("Illegal session factory value. Only [org.springframework.cassandra.core.session.SessionFactory] "
-							+ "and String supported: %s", sessionFactory));
+			throw new IllegalArgumentException(String.format(
+					"Illegal session factory value. Only [org.springframework.cassandra.core.session.SessionFactory]"
+							+ " and String supported: %s", sessionFactory));
 		}
 	}
 

spring-cql/src/main/java/org/springframework/cassandra/core/session/lookup/MapSessionFactoryLookup.java

@@ -16,8 +16,8 @@
 package org.springframework.cassandra.core.session.lookup;
 
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.springframework.cassandra.core.session.SessionFactory;
 import org.springframework.util.Assert;
@@ -34,7 +34,7 @@ import org.springframework.util.Assert;
  */
 public class MapSessionFactoryLookup implements SessionFactoryLookup {
 
-	private final Map<String, SessionFactory> sessionFactories = new HashMap<>(4);
+	private final Map<String, SessionFactory> sessionFactories = new ConcurrentHashMap<>(4);
 
 	/**
 	 * Create a new instance of {@link MapSessionFactoryLookup}.
@@ -110,6 +110,7 @@ public class MapSessionFactoryLookup implements SessionFactoryLookup {
 		Assert.notNull(sessionFactoryName, "SessionFactory name must not be null");
 
 		SessionFactory sessionFactory = this.sessionFactories.get(sessionFactoryName);
+
 		if (sessionFactory == null) {
 			throw new SessionFactoryLookupFailureException(
 					String.format("No SessionFactory with name [%s] registered", sessionFactoryName));

spring-cql/src/main/java/org/springframework/cassandra/support/CassandraAccessor.java

@@ -136,7 +136,9 @@ public class CassandraAccessor implements InitializingBean {
 	 * @see org.springframework.cassandra.support.CassandraExceptionTranslator
 	 */
 	public CassandraExceptionTranslator getExceptionTranslator() {
-		Assert.state(this.exceptionTranslator != null, "CassandraExceptionTranslator was not properly initialized");
+		Assert.state(this.exceptionTranslator != null,
+			"CassandraExceptionTranslator was not properly initialized");
+
 		return this.exceptionTranslator;
 	}
 

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

@@ -15,10 +15,7 @@
  */
 package org.springframework.cassandra.core;
 
-import static org.assertj.core.api.Assertions.*;
-
-import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
+import static org.assertj.core.api.Assertions.assertThat;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -26,6 +23,9 @@ import org.springframework.cassandra.core.session.DefaultBridgedReactiveSession;
 import org.springframework.cassandra.core.session.ReactiveResultSet;
 import org.springframework.cassandra.test.integration.AbstractKeyspaceCreatingIntegrationTest;
 
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
 import com.datastax.driver.core.KeyspaceMetadata;
 import com.datastax.driver.core.PreparedStatement;
 import com.datastax.driver.core.Row;
@@ -33,7 +33,7 @@ import com.datastax.driver.core.exceptions.SyntaxError;
 
 /**
  * Integration tests for {@link DefaultBridgedReactiveSession}.
- * 
+ *
  * @author Mark Paluch
  */
 public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
@@ -51,29 +51,26 @@ public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeysp
 	@Test // DATACASS-335
 	public void executeShouldExecuteDeferred() throws Exception {
 
-		Mono<ReactiveResultSet> execution = reactiveSession
-				.execute("CREATE TABLE users (\n" + "  userid text PRIMARY KEY,\n" + "  first_name text\n" + ");");
+		String query = "CREATE TABLE users (\n" + "  userid text PRIMARY KEY,\n" + "  first_name text\n" + ");";
+
+		Mono<ReactiveResultSet> execution = reactiveSession.execute(query);
 
 		KeyspaceMetadata keyspace = getKeyspaceMetadata();
 
 		assertThat(keyspace.getTable("users")).isNull();
 
 		ReactiveResultSet resultSet = execution.block();
+
 		assertThat(resultSet.wasApplied()).isTrue();
 		assertThat(keyspace.getTable("users")).isNotNull();
 	}
 
-	@Test // DATACASS-335
+	@Test(expected = SyntaxError.class) // DATACASS-335
 	public void executeShouldTransportExceptionsInMono() throws Exception {
 
 		Mono<ReactiveResultSet> execution = reactiveSession.execute("INSERT INTO dummy;");
 
-		try {
-			execution.block();
-			fail("Missing SyntaxError");
-		} catch (SyntaxError e) {
-			assertThat(e).isInstanceOf(SyntaxError.class);
-		}
+		execution.block();
 	}
 
 	@Test // DATACASS-335
@@ -95,8 +92,8 @@ public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeysp
 
 		session.execute("CREATE TABLE users (\n" + "  userid text PRIMARY KEY,\n" + "  first_name text\n" + ");");
 
-		Mono<PreparedStatement> execution = reactiveSession
-				.prepare("INSERT INTO users (userid, first_name) VALUES (?, ?);");
+		Mono<PreparedStatement> execution = reactiveSession.prepare(
+				"INSERT INTO users (userid, first_name) VALUES (?, ?);");
 		PreparedStatement preparedStatement = execution.block();
 
 		assertThat(preparedStatement).isNotNull();

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

@@ -15,10 +15,10 @@
  */
 package org.springframework.cassandra.core;
 
-import static org.assertj.core.api.Assertions.*;
-import static org.mockito.Mockito.*;
-
-import reactor.core.scheduler.Schedulers;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import java.util.Collections;
 
@@ -31,6 +31,8 @@ import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 import org.springframework.cassandra.core.session.DefaultBridgedReactiveSession;
 
+import reactor.core.scheduler.Schedulers;
+
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.Session;
 import com.datastax.driver.core.SimpleStatement;
@@ -38,7 +40,7 @@ import com.datastax.driver.core.Statement;
 
 /**
  * Unit tests for {@link DefaultBridgedReactiveSession}.
- * 
+ *
  * @author Mark Paluch
  */
 @RunWith(MockitoJUnitRunner.class)
@@ -57,6 +59,7 @@ public class DefaultBridgedReactiveSessionUnitTests {
 	public void executeStatementShouldForwardStatementToSession() throws Exception {
 
 		SimpleStatement statement = new SimpleStatement("SELECT *");
+
 		reactiveSession.execute(statement).subscribe();
 
 		verify(sessionMock).executeAsync(statement);
@@ -83,8 +86,8 @@ public class DefaultBridgedReactiveSessionUnitTests {
 
 		reactiveSession.execute("SELECT * WHERE a = ?", Collections.singletonMap("a", "value")).subscribe();
 
-		verify(sessionMock)
-				.executeAsync(eq(new SimpleStatement("SELECT * WHERE a = ?", Collections.singletonMap("a", "value"))));
+		verify(sessionMock).executeAsync(eq(new SimpleStatement("SELECT * WHERE a = ?",
+				Collections.singletonMap("a", "value"))));
 	}
 
 	@Test // DATACASS-335

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

@@ -15,9 +15,7 @@
  */
 package org.springframework.cassandra.core;
 
-import static org.assertj.core.api.Assertions.*;
-
-import reactor.core.scheduler.Schedulers;
+import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -29,11 +27,13 @@ import org.springframework.cassandra.core.session.DefaultReactiveSessionFactory;
 import org.springframework.cassandra.core.session.ReactiveSession;
 import org.springframework.cassandra.test.integration.AbstractKeyspaceCreatingIntegrationTest;
 
+import reactor.core.scheduler.Schedulers;
+
 import com.datastax.driver.core.querybuilder.QueryBuilder;
 
 /**
  * Integration tests for {@link ReactiveCqlTemplate}.
- * 
+ *
  * @author Mark Paluch
  */
 public class ReactiveCqlTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {

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

@@ -15,11 +15,16 @@
  */
 package org.springframework.cassandra.core;
 
-import static org.assertj.core.api.Assertions.*;
-import static org.mockito.Mockito.*;
-
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.atMost;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
 
 import java.util.Collections;
 import java.util.List;
@@ -39,6 +44,9 @@ import org.springframework.cassandra.support.exception.CassandraConnectionFailur
 import org.springframework.cassandra.support.exception.CassandraInvalidQueryException;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import com.datastax.driver.core.BoundStatement;
 import com.datastax.driver.core.ColumnDefinitions;
 import com.datastax.driver.core.ConsistencyLevel;
@@ -52,7 +60,7 @@ import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 
 /**
  * Unit tests for {@link ReactiveCqlTemplate}.
- * 
+ *
  * @author Mark Paluch
  */
 @RunWith(MockitoJUnitRunner.class)

spring-cql/src/test/java/org/springframework/cassandra/core/session/lookup/AbstractRoutingSessionFactoryUnitTests.java

@@ -15,7 +15,8 @@
  */
 package org.springframework.cassandra.core.session.lookup;
 
-import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
 
 import java.util.Collections;
 import java.util.Map;
@@ -46,7 +47,6 @@ public class AbstractRoutingSessionFactoryUnitTests {
 	public void before() throws Exception {
 
 		sut = new StubbedRoutingSessionFactory();
-
 		sut.setDefaultTargetSessionFactory(new DefaultSessionFactory(defaultSession));
 	}
 
@@ -55,7 +55,6 @@ public class AbstractRoutingSessionFactoryUnitTests {
 
 		sut.setTargetSessionFactories(Collections.singletonMap("key", new DefaultSessionFactory(routedSession)));
 		sut.afterPropertiesSet();
-
 		sut.setLookupKey("key");
 
 		assertThat(sut.getSession()).isSameAs(routedSession);
@@ -66,7 +65,6 @@ public class AbstractRoutingSessionFactoryUnitTests {
 
 		sut.setTargetSessionFactories(Collections.singletonMap("key", new DefaultSessionFactory(routedSession)));
 		sut.afterPropertiesSet();
-
 		sut.setLookupKey("unknown");
 
 		assertThat(sut.getSession()).isSameAs(defaultSession);
@@ -76,6 +74,7 @@ public class AbstractRoutingSessionFactoryUnitTests {
 	public void initializationShouldFailUnsupportedLookupKey() {
 
 		sut.setTargetSessionFactories(Collections.singletonMap("key", new Object()));
+
 		try {
 			sut.afterPropertiesSet();
 			fail("Missing IllegalArgumentException");
@@ -98,19 +97,14 @@ public class AbstractRoutingSessionFactoryUnitTests {
 		}
 	}
 
-	@Test // DATACASS-330
+	@Test(expected = IllegalStateException.class) // DATACASS-330
 	public void unresolvableSessionRetrievalShouldFail() {
 
 		sut.setLenientFallback(false);
 		sut.setTargetSessionFactories(Collections.singletonMap("key", new DefaultSessionFactory(routedSession)));
 		sut.afterPropertiesSet();
-
 		sut.setLookupKey("unknown");
-
-		try {
-			sut.getSession();
-			fail("Missing IllegalStateException");
-		} catch (RuntimeException e) {}
+		sut.getSession();
 	}
 
 	@Test // DATACASS-330
@@ -118,7 +112,6 @@ public class AbstractRoutingSessionFactoryUnitTests {
 
 		sut.setTargetSessionFactories(Collections.singletonMap("key", new DefaultSessionFactory(routedSession)));
 		sut.afterPropertiesSet();
-
 		sut.setLookupKey(null);
 
 		assertThat(sut.getSession()).isSameAs(defaultSession);
@@ -147,7 +140,6 @@ public class AbstractRoutingSessionFactoryUnitTests {
 		sut.setSessionFactoryLookup(lookup);
 		sut.setTargetSessionFactories((Map) lookup.getSessionFactories());
 		sut.afterPropertiesSet();
-
 		sut.setLookupKey("lookup-key");
 
 		assertThat(sut.getSession()).isSameAs(defaultSession);

spring-cql/src/test/java/org/springframework/cassandra/core/session/lookup/BeanFactorySessionFactoryLookupUnitTests.java

@@ -48,6 +48,7 @@ public class BeanFactorySessionFactoryLookupUnitTests {
 		when(beanFactory.getBean("factory", SessionFactory.class)).thenReturn(sessionFactory);
 
 		BeanFactorySessionFactoryLookup lookup = new BeanFactorySessionFactoryLookup();
+
 		lookup.setBeanFactory(beanFactory);
 
 		SessionFactory result = lookup.getSessionFactory("factory");
@@ -61,6 +62,7 @@ public class BeanFactorySessionFactoryLookupUnitTests {
 		when(beanFactory.getBean("factory", SessionFactory.class)).thenThrow(new NoSuchBeanDefinitionException("factory"));
 
 		BeanFactorySessionFactoryLookup lookup = new BeanFactorySessionFactoryLookup();
+
 		lookup.setBeanFactory(beanFactory);
 
 		try {

spring-cql/src/test/java/org/springframework/cassandra/core/session/lookup/MapSessionFactoryLookupUnitTests.java

@@ -15,7 +15,8 @@
  */
 package org.springframework.cassandra.core.session.lookup;
 
-import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
 
 import java.util.Collections;
 
@@ -51,7 +52,9 @@ public class MapSessionFactoryLookupUnitTests {
 	@Test // DATACASS-330
 	public void shouldResolveSessionFactoryCorrectly() {
 
-		MapSessionFactoryLookup sessionFactoryLookup = new MapSessionFactoryLookup("factory", sessionFactory);
+		MapSessionFactoryLookup sessionFactoryLookup =
+				new MapSessionFactoryLookup("factory", sessionFactory);
+
 		assertThat(sessionFactoryLookup.getSessionFactory("factory")).isSameAs(sessionFactory);
 	}
 
@@ -60,6 +63,7 @@ public class MapSessionFactoryLookupUnitTests {
 
 		MapSessionFactoryLookup sessionFactoryLookup = new MapSessionFactoryLookup(
 				Collections.singletonMap("factory", sessionFactory));
+
 		assertThat(sessionFactoryLookup.getSessionFactory("factory")).isSameAs(sessionFactory);
 	}
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java

@@ -80,8 +80,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
 	private final CQLExceptionTranslator exceptionTranslator;
 
 	/**
-	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link Session} and a default
-	 * {@link MappingCassandraConverter}.
+	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link Session}
+	 * and a default {@link MappingCassandraConverter}.
 	 *
 	 * @param session {@link Session} used to interact with Cassandra; must not be {@literal null}.
 	 * @see CassandraConverter
@@ -92,12 +92,12 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
 	}
 
 	/**
-	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link Session} and
-	 * {@link CassandraConverter}.
+	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link Session}
+	 * and {@link CassandraConverter}.
 	 *
 	 * @param session {@link Session} used to interact with Cassandra; must not be {@literal null}.
-	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
-	 *          {@literal null}.
+	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types;
+	 * must not be {@literal null}.
 	 * @see CassandraConverter
 	 * @see Session
 	 */
@@ -106,12 +106,12 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
 	}
 
 	/**
-	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link SessionFactory} and
-	 * {@link CassandraConverter}.
+	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link SessionFactory}
+	 * and {@link CassandraConverter}.
 	 *
 	 * @param sessionFactory {@link SessionFactory} used to interact with Cassandra; must not be {@literal null}.
-	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
-	 *          {@literal null}.
+	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types;
+	 * must not be {@literal null}.
 	 * @see CassandraConverter
 	 * @see Session
 	 */
@@ -120,12 +120,12 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
 	}
 
 	/**
-	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link AsyncCqlTemplate} and
-	 * {@link CassandraConverter}.
+	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link AsyncCqlTemplate}
+	 * and {@link CassandraConverter}.
 	 *
 	 * @param asyncCqlTemplate {@link AsyncCqlTemplate} used to interact with Cassandra; must not be {@literal null}.
-	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
-	 *          {@literal null}.
+	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types;
+	 * must not be {@literal null}.
 	 * @see CassandraConverter
 	 * @see Session
 	 */

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraAdminTemplate.java

@@ -132,10 +132,10 @@ public class CassandraAdminTemplate extends CassandraTemplate implements Cassand
 			public KeyspaceMetadata doInSession(Session session) throws DataAccessException {
 
 				KeyspaceMetadata keyspaceMetadata = session.getCluster().getMetadata()
-					.getKeyspace(session.getLoggedKeyspace());
+						.getKeyspace(session.getLoggedKeyspace());
 
-				Assert.state(keyspaceMetadata != null, String.format("Metadata for keyspace [%s] not available",
-					session.getLoggedKeyspace()));
+				Assert.state(keyspaceMetadata != null,
+						String.format("Metadata for keyspace [%s] not available", session.getLoggedKeyspace()));
 
 				return keyspaceMetadata;
 			}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java

@@ -75,8 +75,8 @@ public class CassandraTemplate implements CassandraOperations {
 	private final CqlOperations cqlOperations;
 
 	/**
-	 * Creates an instance of {@link CassandraTemplate} initialized with the given {@link Session} and a default
-	 * {@link MappingCassandraConverter}.
+	 * Creates an instance of {@link CassandraTemplate} initialized with the given {@link Session}
+	 * and a default {@link MappingCassandraConverter}.
 	 *
 	 * @param session {@link Session} used to interact with Cassandra; must not be {@literal null}.
 	 * @see CassandraConverter
@@ -87,12 +87,12 @@ public class CassandraTemplate implements CassandraOperations {
 	}
 
 	/**
-	 * Creates an instance of {@link CassandraTemplate} initialized with the given {@link Session} and
-	 * {@link CassandraConverter}.
+	 * Creates an instance of {@link CassandraTemplate} initialized with the given {@link Session}
+	 * and {@link CassandraConverter}.
 	 *
 	 * @param session {@link Session} used to interact with Cassandra; must not be {@literal null}.
-	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
-	 *          {@literal null}.
+	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types;
+	 * must not be {@literal null}.
 	 * @see CassandraConverter
 	 * @see Session
 	 */
@@ -101,12 +101,12 @@ public class CassandraTemplate implements CassandraOperations {
 	}
 
 	/**
-	 * Creates an instance of {@link CassandraTemplate} initialized with the given {@link SessionFactory} and
-	 * {@link CassandraConverter}.
+	 * Creates an instance of {@link CassandraTemplate} initialized with the given {@link SessionFactory}
+	 * and {@link CassandraConverter}.
 	 *
 	 * @param sessionFactory {@link SessionFactory} used to interact with Cassandra; must not be {@literal null}.
-	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
-	 *          {@literal null}.
+	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types;
+	 * must not be {@literal null}.
 	 * @see CassandraConverter
 	 * @see SessionFactory
 	 */
@@ -115,12 +115,12 @@ public class CassandraTemplate implements CassandraOperations {
 	}
 
 	/**
-	 * Creates an instance of {@link CassandraTemplate} initialized with the given {@link CqlOperations} and
-	 * {@link CassandraConverter}.
+	 * Creates an instance of {@link CassandraTemplate} initialized with the given {@link CqlOperations}
+	 * and {@link CassandraConverter}.
 	 *
 	 * @param cqlOperations {@link CqlOperations} used to interact with Cassandra; must not be {@literal null}.
-	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
-	 *          {@literal null}.
+	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types;
+	 * must not be {@literal null}.
 	 * @see CassandraConverter
 	 * @see Session
 	 */
@@ -137,6 +137,7 @@ public class CassandraTemplate implements CassandraOperations {
 	private static MappingCassandraConverter newConverter() {
 
 		MappingCassandraConverter converter = new MappingCassandraConverter();
+
 		converter.afterPropertiesSet();
 
 		return converter;
@@ -323,7 +324,8 @@ public class CassandraTemplate implements CassandraOperations {
 
 		Assert.notNull(entity, "Entity must not be null");
 
-		Insert insert = QueryUtils.createInsertQuery(getTableName(entity.getClass()).toCql(), entity, options, converter);
+		Insert insert = QueryUtils.createInsertQuery(getTableName(entity.getClass()).toCql(),
+				entity, options, converter);
 
 		return cqlOperations.execute(new StatementCallback<>(insert, entity));
 	}
@@ -346,7 +348,8 @@ public class CassandraTemplate implements CassandraOperations {
 
 		Assert.notNull(entity, "Entity must not be null");
 
-		Update update = QueryUtils.createUpdateQuery(getTableName(entity.getClass()).toCql(), entity, options, converter);
+		Update update = QueryUtils.createUpdateQuery(getTableName(entity.getClass()).toCql(),
+				entity, options, converter);
 
 		return cqlOperations.execute(new StatementCallback<>(update, entity));
 	}
@@ -369,7 +372,8 @@ public class CassandraTemplate implements CassandraOperations {
 
 		Assert.notNull(entity, "Entity must not be null");
 
-		Delete delete = QueryUtils.createDeleteQuery(getTableName(entity.getClass()).toCql(), entity, options, converter);
+		Delete delete = QueryUtils.createDeleteQuery(getTableName(entity.getClass()).toCql(),
+				entity, options, converter);
 
 		return cqlOperations.execute(new StatementCallback<>(delete, entity));
 	}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java

@@ -15,10 +15,7 @@
  */
 package org.springframework.data.cassandra.core;
 
-import static org.assertj.core.api.Assertions.*;
-
-import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
+import static org.assertj.core.api.Assertions.assertThat;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -29,6 +26,9 @@ import org.springframework.data.cassandra.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.domain.Person;
 import org.springframework.data.cassandra.test.integration.support.SchemaTestUtils;
 
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
 /**
  * Integration tests for {@link ReactiveCassandraTemplate}.
  *
@@ -42,9 +42,9 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC
 	public void setUp() throws Exception {
 
 		MappingCassandraConverter converter = new MappingCassandraConverter();
-		CassandraTemplate cassandraTemplate = new CassandraTemplate(session, converter);
-
+		CassandraTemplate cassandraTemplate = new CassandraTemplate(this.session, converter);
 		DefaultBridgedReactiveSession session = new DefaultBridgedReactiveSession(this.session, Schedulers.elastic());
+
 		template = new ReactiveCassandraTemplate(new ReactiveCqlTemplate(session), converter);
 
 		SchemaTestUtils.potentiallyCreateTableFor(Person.class, cassandraTemplate);
@@ -57,11 +57,12 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC
 		Person person = new Person("heisenberg", "Walter", "White");
 
 		Mono<Person> insert = template.insert(person);
-
 		Mono<Person> oneById = template.selectOneById(person.getId(), Person.class);
+
 		assertThat(oneById.hasElement().block()).isFalse();
 
 		Person saved = insert.block();
+
 		assertThat(saved).isNotNull().isEqualTo(person);
 		assertThat(oneById.block()).isNotNull().isEqualTo(saved);
 	}
@@ -74,6 +75,7 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC
 		template.insert(person).block();
 
 		Mono<Long> count = template.count(Person.class);
+
 		assertThat(count.block()).isEqualTo(1L);
 	}
 
@@ -81,13 +83,17 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC
 	public void updateShouldUpdateEntity() {
 
 		Person person = new Person("heisenberg", "Walter", "White");
+
 		template.insert(person).block();
 
 		person.setFirstname("Walter Hartwell");
+
 		Person updated = template.update(person).block();
+
 		assertThat(updated).isNotNull();
 
 		Mono<Person> oneById = template.selectOneById(person.getId(), Person.class);
+
 		assertThat(oneById.block()).isEqualTo(person);
 	}
 
@@ -95,12 +101,15 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC
 	public void deleteShouldRemoveEntity() {
 
 		Person person = new Person("heisenberg", "Walter", "White");
+
 		template.insert(person).block();
 
 		Person deleted = template.delete(person).block();
+
 		assertThat(deleted).isNotNull();
 
 		Mono<Person> oneById = template.selectOneById(person.getId(), Person.class);
+
 		assertThat(oneById.block()).isNull();
 	}
 
@@ -108,12 +117,15 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC
 	public void deleteByIdShouldRemoveEntity() {
 
 		Person person = new Person("heisenberg", "Walter", "White");
+
 		template.insert(person).block();
 
 		Boolean deleted = template.deleteById(person.getId(), Person.class).block();
+
 		assertThat(deleted).isTrue();
 
 		Mono<Person> oneById = template.selectOneById(person.getId(), Person.class);
+
 		assertThat(oneById.block()).isNull();
 	}
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateUnitTests.java

@@ -15,14 +15,14 @@
  */
 package org.springframework.data.cassandra.core;
 
-import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
-import static org.mockito.Mockito.*;
 import static org.mockito.Mockito.anyInt;
-
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import java.util.Collections;
 
@@ -38,6 +38,9 @@ import org.springframework.cassandra.core.session.ReactiveSession;
 import org.springframework.cassandra.support.exception.CassandraConnectionFailureException;
 import org.springframework.data.cassandra.domain.Person;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import com.datastax.driver.core.ColumnDefinitions;
 import com.datastax.driver.core.DataType;
 import com.datastax.driver.core.Row;
@@ -46,7 +49,7 @@ import com.datastax.driver.core.exceptions.NoHostAvailableException;
 
 /**
  * Unit tests for {@link ReactiveCassandraTemplate}.
- * 
+ *
  * @author Mark Paluch
  */
 @RunWith(MockitoJUnitRunner.class)
@@ -64,6 +67,7 @@ public class ReactiveCassandraTemplateUnitTests {
 	public void setUp() {
 
 		template = new ReactiveCassandraTemplate(session);
+
 		when(session.execute(anyString())).thenReturn(Mono.just(reactiveResultSet));
 		when(session.execute(any(Statement.class))).thenReturn(Mono.just(reactiveResultSet));
 		when(reactiveResultSet.getColumnDefinitions()).thenReturn(columnDefinitions);

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoriesRegistrarUnitTests.java

@@ -15,7 +15,7 @@
  */
 package org.springframework.data.cassandra.repository.config;
 
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.mock;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -54,11 +54,11 @@ public class ReactiveCassandraRepositoriesRegistrarUnitTests {
 		}
 	}
 
-	@Autowired ReactivePersonRepository personRepository;
 	@Autowired ApplicationContext context;
+	@Autowired ReactivePersonRepository personRepository;
 
 	@Test // DATACASS-335
 	public void testConfiguration() {}
 
-	static interface ReactivePersonRepository extends ReactiveCassandraRepository<Person, String> {}
+	interface ReactivePersonRepository extends ReactiveCassandraRepository<Person, String> {}
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQueryUnitTests.java

@@ -15,8 +15,9 @@
  */
 package org.springframework.data.cassandra.repository.query;
 
-import static org.assertj.core.api.Assertions.*;
-import static org.mockito.Mockito.*;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.when;
 
 import java.lang.reflect.Method;
 
@@ -99,6 +100,7 @@ public class ReactiveStringBasedCassandraQueryUnitTests {
 
 		String table = Person.class.getSimpleName().toLowerCase();
 		Select expected = QueryBuilder.select().all().from(table);
+
 		expected.setForceNoValues(true);
 		expected.where(QueryBuilder.eq("lastname", "White"));
 
@@ -110,6 +112,7 @@ public class ReactiveStringBasedCassandraQueryUnitTests {
 		Method method = ReflectionUtils.findMethod(SampleRepository.class, name, args);
 		ReactiveCassandraQueryMethod queryMethod = new ReactiveCassandraQueryMethod(method, metadata, factory,
 				converter.getMappingContext());
+
 		return new ReactiveStringBasedCassandraQuery(queryMethod, operations, PARSER,
 				new ExtensionAwareEvaluationContextProvider());
 	}