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

DATACASS-126 - Added DataStax Java Driver 2.0.1 dependency and

Browse Source 
cassandra-unit 2.0.X dependency.

Full support for Cassandra 2.0.X
This commit is contained in:
David T Webb
2014-04-29 09:45:14 -04:00
parent b8cbd20ae1
commit 8edc4ce1a9
25 changed files with 246 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-cql/src/main/java/org/springframework/cassandra/config/CassandraCqlClusterFactoryBean.java
View File

@@ -86,7 +86,6 @@ public class CassandraCqlClusterFactoryBean implements FactoryBean<Cluster>, Ini
private ReconnectionPolicy reconnectionPolicy;
private RetryPolicy retryPolicy;
private boolean metricsEnabled = DEFAULT_METRICS_ENABLED;
private boolean deferredInitialization = DEFAULT_DEFERRED_INITIALIZATION;
private boolean jmxReportingEnabled = DEFAULT_JMX_REPORTING_ENABLED;
private boolean sslEnabled = DEFAULT_SSL_ENABLED;
private SSLOptions sslOptions;
@@ -163,10 +162,6 @@ public class CassandraCqlClusterFactoryBean implements FactoryBean<Cluster>, Ini
builder.withRetryPolicy(retryPolicy);
}
if (deferredInitialization) {
builder.withDeferredInitialization();
}
if (!metricsEnabled) {
builder.withoutMetrics();
}
@@ -260,7 +255,7 @@ public class CassandraCqlClusterFactoryBean implements FactoryBean<Cluster>, Ini
} finally {
if (system != null) {
system.shutdown();
system.close();
}
}
}
@@ -269,7 +264,7 @@ public class CassandraCqlClusterFactoryBean implements FactoryBean<Cluster>, Ini
public void destroy() throws Exception {
executeSpecsAndScripts(keyspaceDrops, shutdownScripts);
cluster.shutdown();
cluster.close();
}
/**
@@ -379,13 +374,6 @@ public class CassandraCqlClusterFactoryBean implements FactoryBean<Cluster>, Ini
this.password = password;
}
/**
* @param deferredInitialization The deferredInitialization to set.
*/
public void setDeferredInitialization(boolean deferredInitialization) {
this.deferredInitialization = deferredInitialization;
}
/**
* @param jmxReportingEnabled The jmxReportingEnabled to set.
*/

2
spring-cql/src/main/java/org/springframework/cassandra/config/CassandraCqlSessionFactoryBean.java
View File

@@ -108,7 +108,7 @@ public class CassandraCqlSessionFactoryBean implements FactoryBean<Session>, Ini
public void destroy() throws Exception {
executeScripts(shutdownScripts);
session.shutdown();
session.close();
}
/**

11
spring-cql/src/main/java/org/springframework/cassandra/config/xml/CassandraCqlClusterParser.java
View File

@@ -15,11 +15,6 @@
*/
package org.springframework.cassandra.config.xml;
import static org.springframework.cassandra.config.xml.ParsingUtils.addOptionalPropertyReference;
import static org.springframework.cassandra.config.xml.ParsingUtils.addOptionalPropertyValue;
import static org.springframework.cassandra.config.xml.ParsingUtils.addRequiredPropertyValue;
import static org.springframework.cassandra.config.xml.ParsingUtils.getSourceBeanDefinition;
import java.util.ArrayList;
import java.util.List;
@@ -46,6 +41,11 @@ import com.datastax.driver.core.HostDistance;
import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.SocketOptions;
import static org.springframework.cassandra.config.xml.ParsingUtils.addOptionalPropertyReference;
import static org.springframework.cassandra.config.xml.ParsingUtils.addOptionalPropertyValue;
import static org.springframework.cassandra.config.xml.ParsingUtils.addRequiredPropertyValue;
import static org.springframework.cassandra.config.xml.ParsingUtils.getSourceBeanDefinition;
/**
* Parses the {@literal <cluster>} element of the XML Configuration.
*
@@ -97,7 +97,6 @@ public class CassandraCqlClusterParser extends AbstractBeanDefinitionParser {
addOptionalPropertyValue(builder, "compressionType", element, "compression", null);
addOptionalPropertyValue(builder, "username", element, "username", null);
addOptionalPropertyValue(builder, "password", element, "password", null);
addOptionalPropertyValue(builder, "deferredInitialization", element, "deferred-initialization", null);
addOptionalPropertyValue(builder, "metricsEnabled", element, "metrics-enabled", null);
addOptionalPropertyValue(builder, "jmxReportingEnabled", element, "jmx-reporting-enabled", null);
addOptionalPropertyValue(builder, "sslEnabled", element, "ssl-enabled", null);

28
spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java
View File

@@ -32,10 +32,10 @@ import org.springframework.cassandra.core.keyspace.DropKeyspaceSpecification;
import org.springframework.cassandra.core.keyspace.DropTableSpecification;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.Query;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.querybuilder.Batch;
import com.datastax.driver.core.querybuilder.Delete;
import com.datastax.driver.core.querybuilder.Insert;
@@ -78,9 +78,9 @@ public interface CqlOperations {
/**
* Executes the supplied Query and returns nothing.
*
* @param query The {@link Query} to execute
* @param query The {@link Statement} to execute
*/
void execute(Query query) throws DataAccessException;
void execute(Statement query) throws DataAccessException;
/**
* Executes the supplied Delete Query and returns nothing.
@@ -206,44 +206,44 @@ public interface CqlOperations {
/**
* Executes the supplied CQL Batch Asynchronously and returns nothing.
*
* @param query The {@link Query} to execute
* @param query The {@link Batch} to execute
*/
void executeAsynchronously(Batch batch) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Query} to execute
* @param query The {@link Statement} to execute
*/
void executeAsynchronously(Query query) throws DataAccessException;
void executeAsynchronously(Statement query) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Query} to execute
* @param query The {@link Statement} to execute
*/
void executeAsynchronously(Query query, Runnable runnable) throws DataAccessException;
void executeAsynchronously(Statement query, Runnable runnable) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Query} to execute
* @param query The {@link Statement} to execute
*/
void executeAsynchronously(Query query, AsynchronousQueryListener listener) throws DataAccessException;
void executeAsynchronously(Statement query, AsynchronousQueryListener listener) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Query} to execute
* @param query The {@link Statement} to execute
*/
void executeAsynchronously(Query query, Runnable runnable, Executor executor) throws DataAccessException;
void executeAsynchronously(Statement query, Runnable runnable, Executor executor) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Query} to execute
* @param query The {@link Statement} to execute
*/
void executeAsynchronously(Query query, AsynchronousQueryListener listener, Executor executor)
void executeAsynchronously(Statement query, AsynchronousQueryListener listener, Executor executor)
throws DataAccessException;
/**

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

@@ -57,7 +57,6 @@ import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.ColumnDefinitions.Definition;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Query;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Row;
@@ -94,13 +93,13 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
protected static final Logger log = LoggerFactory.getLogger(CqlTemplate.class);
/**
* Add common {@link Query} options for all types of queries.
* Add common {@link Statement} options for all types of queries.
*
* @param q
* @param options
* @return the {@link Query} given.
* @return the {@link Statement} given.
*/
public static Query addQueryOptions(Query q, QueryOptions options) {
public static Statement addQueryOptions(Statement q, QueryOptions options) {
if (options == null) {
return q;
@@ -224,7 +223,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void execute(Query query) throws DataAccessException {
public void execute(Statement query) throws DataAccessException {
doExecute(query);
}
@@ -505,7 +504,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @param q The query to execute.
* @param options The {@link QueryOptions}. May be null.
*/
protected ResultSet doExecute(final Query q) {
protected ResultSet doExecute(final Statement q) {
return doExecute(new SessionCallback<ResultSet>() {
@@ -521,7 +520,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
});
}
protected ResultSetFuture doExecuteAsync(final Query q) {
protected ResultSetFuture doExecuteAsync(final Statement q) {
return doExecute(new SessionCallback<ResultSetFuture>() {
@@ -669,12 +668,12 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void executeAsynchronously(Query query) throws DataAccessException {
public void executeAsynchronously(Statement query) throws DataAccessException {
doExecuteAsync(query);
}
@Override
public void executeAsynchronously(Query query, Runnable listener) throws DataAccessException {
public void executeAsynchronously(Statement query, Runnable listener) throws DataAccessException {
executeAsynchronously(query, listener, new Executor() {
@Override
@@ -685,7 +684,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void executeAsynchronously(Query query, AsynchronousQueryListener listener) throws DataAccessException {
public void executeAsynchronously(Statement query, AsynchronousQueryListener listener) throws DataAccessException {
executeAsynchronously(query, listener, new Executor() {
@Override
@@ -696,7 +695,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void executeAsynchronously(final Query query, final Runnable listener, final Executor executor)
public void executeAsynchronously(final Statement query, final Runnable listener, final Executor executor)
throws DataAccessException {
execute(new SessionCallback<Object>() {
@Override
@@ -709,8 +708,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void executeAsynchronously(final Query query, final AsynchronousQueryListener listener, final Executor executor)
throws DataAccessException {
public void executeAsynchronously(final Statement query, final AsynchronousQueryListener listener,
final Executor executor) throws DataAccessException {
execute(new SessionCallback<Object>() {
@Override
@@ -991,7 +990,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
@Override
public void truncate(CqlIdentifier tableName) throws DataAccessException {
Truncate truncate = QueryBuilder.truncate(tableName.toCql());
doExecute(truncate.getQueryString(), null);
doExecute(truncate);
}
@Override
@@ -1205,7 +1204,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
@Override
public long count(CqlIdentifier tableName) {
return selectCount(QueryBuilder.select().countAll().from(tableName.toCql()).getQueryString());
return selectCount(QueryBuilder.select().countAll().from(tableName.toCql()));
}
@Override
@@ -1213,9 +1212,9 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
return count(cqlId(tableName));
}
protected long selectCount(String countQuery) {
protected long selectCount(Select select) {
return query(countQuery, new ResultSetExtractor<Long>() {
return query(select, new ResultSetExtractor<Long>() {
@Override
public Long extractData(ResultSet rs) throws DriverException, DataAccessException {

8
spring-cql/src/main/java/org/springframework/cassandra/support/exception/CassandraConnectionFailureException.java
View File

@@ -31,14 +31,14 @@ public class CassandraConnectionFailureException extends DataAccessResourceFailu
private static final long serialVersionUID = 6299912054261646552L;
private final Map<InetAddress, String> messagesByHost = new HashMap<InetAddress, String>();
private final Map<InetAddress, Throwable> messagesByHost = new HashMap<InetAddress, Throwable>();
public CassandraConnectionFailureException(Map<InetAddress, String> messagesByHost, String msg, Throwable cause) {
public CassandraConnectionFailureException(Map<InetAddress, Throwable> map, String msg, Throwable cause) {
super(msg, cause);
this.messagesByHost.putAll(messagesByHost);
this.messagesByHost.putAll(map);
}
public Map<InetAddress, String> getMessagesByHost() {
public Map<InetAddress, Throwable> getMessagesByHost() {
return Collections.unmodifiableMap(messagesByHost);
}
}

8
spring-cql/src/main/resources/org/springframework/cassandra/config/spring-cql-1.0.xsd
View File

@@ -172,14 +172,6 @@ Determine whether or not to enable JMX Reporting. Defaults to true.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="deferred-initialization" type="xsd:string"
default="false">
<xsd:annotation>
<xsd:documentation><![CDATA[
Determine if we defer initalizing the cluster until a connection is requested. Defaults to false.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="ssl-enabled" type="xsd:string"
default="false">
<xsd:annotation>

4
spring-cql/src/test/java/org/springframework/cassandra/test/integration/AbstractEmbeddedCassandraIntegrationTest.java
View File

@@ -63,8 +63,8 @@ public class AbstractEmbeddedCassandraIntegrationTest {
}
@BeforeClass
public static void startCassandra() throws ConfigurationException, TTransportException, IOException,
InterruptedException {
public static void startCassandra() throws TTransportException, IOException, InterruptedException,
ConfigurationException {
EmbeddedCassandraServerHelper.startEmbeddedCassandra(CASSANDRA_CONFIG);
}

8
spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/java/AbstractKeyspaceCreatingConfiguration.java
View File

@@ -15,9 +15,6 @@
*/
package org.springframework.cassandra.test.integration.config.java;
import static org.springframework.cassandra.core.cql.generator.CreateKeyspaceCqlGenerator.toCql;
import static org.springframework.cassandra.core.keyspace.CreateKeyspaceSpecification.createKeyspace;
import org.springframework.cassandra.config.CassandraCqlSessionFactoryBean;
import org.springframework.cassandra.config.java.AbstractSessionConfiguration;
import org.springframework.context.annotation.Configuration;
@@ -26,6 +23,9 @@ import org.springframework.util.StringUtils;
import com.datastax.driver.core.KeyspaceMetadata;
import com.datastax.driver.core.Session;
import static org.springframework.cassandra.core.cql.generator.CreateKeyspaceCqlGenerator.toCql;
import static org.springframework.cassandra.core.keyspace.CreateKeyspaceSpecification.createKeyspace;
@Configuration
public abstract class AbstractKeyspaceCreatingConfiguration extends AbstractSessionConfiguration {
@@ -50,6 +50,6 @@ public abstract class AbstractKeyspaceCreatingConfiguration extends AbstractSess
}
system.execute(toCql(createKeyspace().name(keyspace).withSimpleReplication()));
system.shutdown();
system.close();
}
}

3
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CQLOperationsTest.java
View File

@@ -29,7 +29,6 @@ import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.mortbay.log.Log;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cassandra.core.ConsistencyLevel;
@@ -1145,7 +1144,7 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
@Test
public void insertAndTruncateQueryObjectTest() {
Log.info("Starting Insert and Truncate Query Object Test");
log.info("Starting Insert and Truncate Query Object Test");
String tableName = "truncate_test";

8
spring-cql/src/test/resources/org/springframework/cassandra/test/integration/config/xml/PropertyPlaceholderNamespaceCreatingXmlConfigTest-context.xml
View File

@@ -9,14 +9,16 @@
<context:property-placeholder
location="classpath:/org/springframework/cassandra/test/integration/config/xml/ppncxct.properties" />
<bean id="authProvider" class="com.datastax.driver.core.sasl.DseAuthProvider" />
<bean id="authProvider" class="com.datastax.driver.auth.DseAuthProvider" />
<!--
<bean id="loadBalancingPolicy"
class="com.datastax.driver.core.policies.DCAwareRoundRobinPolicy">
<constructor-arg name="localDc" value="${lb.policy.dcAware.localDc}" />
<constructor-arg name="usedHostsPerRemoteDc"
value="${lb.policy.dcAware.remoteHosts}" />
</bean>
-->
<bean id="reconnectionPolicy"
class="com.datastax.driver.core.policies.ConstantReconnectionPolicy">
@@ -35,9 +37,9 @@
<cassandra:cluster contact-points="${cluster.contactPoints}"
port="${cluster.port}" compression="${cluster.compression}"
auth-info-provider-ref="authProvider" load-balancing-policy-ref="loadBalancingPolicy"
auth-info-provider-ref="authProvider"
username="${auth.username}" password="${auth.password}"
deferred-initialization="${cluster.deferredInit}" metrics-enabled="${cluster.metricsEnabled}"
metrics-enabled="${cluster.metricsEnabled}"
jmx-reporting-enabled="${cluster.jmxReportingEnabled}"
reconnection-policy-ref="reconnectionPolicy" retry-policy-ref="retryPolicy"
host-state-listener-ref="hostStateListener" latency-tracker-ref="latencyTracker">

3
spring-cql/src/test/resources/org/springframework/cassandra/test/integration/config/xml/ppncxct.properties
View File

@@ -1,7 +1,6 @@
cluster.contactPoints=localhost
cluster.port=@build.cassandra.native_transport_port@
cluster.compression=SNAPPY
cluster.deferredInit=true
cluster.metricsEnabled=false
cluster.jmxReportingEnabled=false
cluster.reconnection.delayMillis=5000
@@ -12,7 +11,7 @@ dc1.name=DCJAX
dc1.rf=2
dc2.name=DCCTL
dc2.rf=3
lb.policy.dcAware.remoteHosts=4
lb.policy.dcAware.remoteHosts=0
lb.policy.dcAware.localDc=DCJAX
auth.username=test
auth.password=pass

44
spring-cql/src/test/resources/spring-cassandra.yaml
View File

@@ -1,6 +1,5 @@
# Cassandra storage config YAML
# NOTE:
# See http://wiki.apache.org/cassandra/StorageConfiguration for
# full explanations of configuration directives
# /NOTE
@@ -170,24 +169,6 @@ row_cache_save_period: 0
# Disabled by default, meaning all keys are going to be saved
# row_cache_keys_to_save: 100
# The provider for the row cache to use.
#
# Supported values are: ConcurrentLinkedHashCacheProvider, SerializingCacheProvider
#
# SerializingCacheProvider serialises the contents of the row and stores
# it in native memory, i.e., off the JVM Heap. Serialized rows take
# significantly less memory than "live" rows in the JVM, so you can cache
# more rows in a given memory footprint. And storing the cache off-heap
# means you can use smaller heap sizes, reducing the impact of GC pauses.
# Note however that when a row is requested from the row cache, it must be
# deserialized into the heap for use.
#
# It is also valid to specify the fully-qualified class name to a class
# that implements org.apache.cassandra.cache.IRowCacheProvider.
#
# Defaults to SerializingCacheProvider
row_cache_provider: SerializingCacheProvider
# saved caches
saved_caches_directory: target/embeddedCassandra/saved_caches
@@ -230,31 +211,6 @@ seed_provider:
# Ex: "<ip1>,<ip2>,<ip3>"
- seeds: "127.0.0.1"
# emergency pressure valve: each time heap usage after a full (CMS)
# garbage collection is above this fraction of the max, Cassandra will
# flush the largest memtables.
#
# Set to 1.0 to disable. Setting this lower than
# CMSInitiatingOccupancyFraction is not likely to be useful.
#
# RELYING ON THIS AS YOUR PRIMARY TUNING MECHANISM WILL WORK POORLY:
# it is most effective under light to moderate load, or read-heavy
# workloads; under truly massive write load, it will often be too
# little, too late.
flush_largest_memtables_at: 0.75
# emergency pressure valve #2: the first time heap usage after a full
# (CMS) garbage collection is above this fraction of the max,
# Cassandra will reduce cache maximum _capacity_ to the given fraction
# of the current _size_. Should usually be set substantially above
# flush_largest_memtables_at, since that will have less long-term
# impact on the system.
#
# Set to 1.0 to disable. Setting this lower than
# CMSInitiatingOccupancyFraction is not likely to be useful.
reduce_cache_sizes_at: 0.85
reduce_cache_capacity_to: 0.6
# For workloads with more data than can fit in memory, Cassandra's
# bottleneck will be reads that need to fetch data from
# disk. "concurrent_reads" should be set to (16 * number_of_drives) in