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

DATACASS-271 - Upgrade to Cassandra 3.0.

Browse Source 
This commit adds support for Cassandra 3.x with upgrading to Datastax Driver 3.0.0. Data type deserialization is performed using the CodecRegistry.

Original pull request: #52.
Related ticket: DATACASS-169.
This commit is contained in:
Antoine Toulme
2016-04-27 23:57:09 -07:00
committed by John Blum
parent 27019f3218
commit a920b530df
18 changed files with 99 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
pom.xml
View File

@@ -21,6 +21,25 @@
<description>Spring Data Cassandra</description>
<url>http://projects.spring.io/spring-data-cassandra/</url>
<properties>
<project.type>multi</project.type>
<dist.id>spring-data-cassandra</dist.id>
<springdata.commons>1.13.0.BUILD-SNAPSHOT</springdata.commons>
<cassandra-unit.version>3.0.0.1</cassandra-unit.version>
<el.version>1.0</el.version>
<failsafe.version>2.16</failsafe.version>
<jamm.version>0.3.1</jamm.version>
<cassandra.version>3.0.0</cassandra.version>
<cassandra-driver-dse.version>3.0.0-rc1</cassandra-driver-dse.version>
<build.cassandra.mode>embedded</build.cassandra.mode>
<build.cassandra.host>localhost</build.cassandra.host>
<build.cassandra.native_transport_port>19042</build.cassandra.native_transport_port>
<build.cassandra.rpc_port>19160</build.cassandra.rpc_port>
<build.cassandra.storage_port>17000</build.cassandra.storage_port>
<build.cassandra.ssl_storage_port>17001</build.cassandra.ssl_storage_port>
</properties>
<developers>
<developer>
<id>madams</id>
@@ -64,23 +83,6 @@
<module>spring-data-cassandra-distribution</module>
</modules>
<properties>
<build.cassandra.host>localhost</build.cassandra.host>
<build.cassandra.mode>embedded</build.cassandra.mode>
<build.cassandra.native_transport_port>19042</build.cassandra.native_transport_port>
<build.cassandra.rpc_port>19160</build.cassandra.rpc_port>
<build.cassandra.ssl_storage_port>17001</build.cassandra.ssl_storage_port>
<build.cassandra.storage_port>17000</build.cassandra.storage_port>
<cassandra.version>2.1.11</cassandra.version>
<cassandra-driver-dse.version>2.1.7.1</cassandra-driver-dse.version>
<cassandra-unit.version>2.1.9.2</cassandra-unit.version>
<dist.id>spring-data-cassandra</dist.id>
<el.version>1.0</el.version>
<failsafe.version>2.16</failsafe.version>
<project.type>multi</project.type>
<springdata.commons>1.13.0.BUILD-SNAPSHOT</springdata.commons>
</properties>
<repositories>
<repository>
<id>spring-libs-snapshot</id>
@@ -141,7 +143,7 @@
<dependency>
<groupId>org.xerial.snappy</groupId>
<artifactId>snappy-java</artifactId>
<version>1.1.0.1</version>
<version>1.1.2.1</version>
<scope>test</scope>
</dependency>

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

@@ -55,6 +55,7 @@ import org.springframework.dao.QueryTimeoutException;
import org.springframework.util.Assert;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.ColumnDefinitions.Definition;
import com.datastax.driver.core.Host;
@@ -571,7 +572,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
if (cols.size() == 0) {
return null;
}
return cols.getType(0).deserialize(row.getBytesUnsafe(0), ProtocolVersion.NEWEST_SUPPORTED);
return CodecRegistry.DEFAULT_INSTANCE.codecFor(cols.getType(0)).deserialize(row.getBytesUnsafe(0), ProtocolVersion.NEWEST_SUPPORTED);
}
/**
@@ -588,7 +589,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
for (Definition def : cols.asList()) {
String name = def.getName();
map.put(name, def.getType().deserialize(row.getBytesUnsafe(name), ProtocolVersion.NEWEST_SUPPORTED));
map.put(name, CodecRegistry.DEFAULT_INSTANCE.codecFor(def.getType()).deserialize(row.getBytesUnsafe(name), ProtocolVersion.NEWEST_SUPPORTED));
}
return map;

5
spring-cql/src/main/java/org/springframework/cassandra/core/converter/RowToListConverter.java
View File

@@ -22,6 +22,7 @@ import java.util.List;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.convert.ReadingConverter;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.ColumnDefinitions.Definition;
import com.datastax.driver.core.ProtocolVersion;
@@ -51,8 +52,8 @@ public class RowToListConverter implements Converter<Row, List<Object>> {
for (Definition def : cols.asList()) {
String name = def.getName();
list.add(row.isNull(name) ? null
: def.getType().deserialize(row.getBytesUnsafe(name), ProtocolVersion.NEWEST_SUPPORTED));
list.add(row.isNull(name) ? null : CodecRegistry.DEFAULT_INSTANCE.codecFor(def.getType()).deserialize(
row.getBytesUnsafe(name), ProtocolVersion.NEWEST_SUPPORTED));
}
return list;

7
spring-cql/src/main/java/org/springframework/cassandra/core/converter/RowToMapConverter.java
View File

@@ -22,6 +22,7 @@ import java.util.Map;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.convert.ReadingConverter;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.ColumnDefinitions.Definition;
import com.datastax.driver.core.ProtocolVersion;
@@ -52,8 +53,10 @@ public class RowToMapConverter implements Converter<Row, Map<String, Object>> {
for (Definition def : cols.asList()) {
String name = def.getName();
map.put(name, row.isNull(name) ? null
: def.getType().deserialize(row.getBytesUnsafe(name), ProtocolVersion.NEWEST_SUPPORTED));
map.put(
name,
row.isNull(name) ? null : CodecRegistry.DEFAULT_INSTANCE.codecFor(def.getType())
.deserialize(row.getBytesUnsafe(name), ProtocolVersion.NEWEST_SUPPORTED));
}
return map;

2
spring-cql/src/test/java/org/springframework/cassandra/test/integration/EmbeddedCassandraServerHelper.java
View File

@@ -193,7 +193,7 @@ class EmbeddedCassandraServerHelper {
CommitLog commitLog = CommitLog.instance;
commitLog.getContext(); // wait for commit log allocator instantiation to avoid hanging on a race condition
commitLog.resetUnsafe(); // cleanup screws w/ CommitLog, this brings it back to safe state
commitLog.resetUnsafe(true); // cleanup screws w/ CommitLog, this brings it back to safe state
}
private static void cleanup() throws IOException {

6
spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/CassandraCqlClusterFactoryBeanIntegrationTests.java
View File

@@ -47,11 +47,11 @@ public class CassandraCqlClusterFactoryBeanIntegrationTests extends AbstractEmbe
@Test
public void configuredProtocolVersionShouldBeSet() throws Exception {
cassandraCqlClusterFactoryBean.setProtocolVersion(ProtocolVersion.V2);
cassandraCqlClusterFactoryBean.setProtocolVersion(ProtocolVersion.V4);
cassandraCqlClusterFactoryBean.setPort(cassandraEnvironment.getPort());
cassandraCqlClusterFactoryBean.afterPropertiesSet();
assertEquals(ProtocolVersion.V2, getProtocolVersionEnum(cassandraCqlClusterFactoryBean));
assertEquals(ProtocolVersion.V4, getProtocolVersionEnum(cassandraCqlClusterFactoryBean));
}
@Test
@@ -68,6 +68,6 @@ public class CassandraCqlClusterFactoryBeanIntegrationTests extends AbstractEmbe
// initialize connection factory
cassandraCqlClusterFactoryBean.getObject().init();
return cassandraCqlClusterFactoryBean.getObject().getConfiguration().getProtocolOptions().getProtocolVersionEnum();
return cassandraCqlClusterFactoryBean.getObject().getConfiguration().getProtocolOptions().getProtocolVersion();
}
}

10
spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/xml/TestHostStateListener.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cassandra.test.integration.config.xml;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Host.StateListener;
@@ -49,8 +50,13 @@ public class TestHostStateListener implements StateListener {
}
@Override
public void onSuspected(Host host) {
log.info("Host Suspected: " + host.getAddress());
public void onRegister(Cluster cluster) {
log.info("Cluster registered: " + cluster.getClusterName());
}
@Override
public void onUnregister(Cluster cluster) {
log.info("Cluster unregistered: " + cluster.getClusterName());
}
}

9
spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/xml/TestLatencyTracker.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cassandra.test.integration.config.xml;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.LatencyTracker;
import com.datastax.driver.core.Statement;
@@ -34,5 +35,13 @@ public class TestLatencyTracker implements LatencyTracker {
public void update(Host host, Statement statement, Exception exception, long newLatencyNanos) {
LOG.info("Latency Tracker: " + host.getAddress() + ", " + newLatencyNanos + " nanoseconds.");
}
@Override
public void onRegister(Cluster cluster) {
}
@Override
public void onUnregister(Cluster cluster) {
}
}

17
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/cql/generator/CqlIndexSpecificationAssertions.java
View File

@@ -19,8 +19,9 @@ import static org.junit.Assert.*;
import org.springframework.cassandra.core.keyspace.IndexDescriptor;
import com.datastax.driver.core.ColumnMetadata.IndexMetadata;
import com.datastax.driver.core.IndexMetadata;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.TableMetadata;
/**
* @author David Webb
@@ -29,15 +30,19 @@ import com.datastax.driver.core.Session;
public class CqlIndexSpecificationAssertions {
public static void assertIndex(IndexDescriptor expected, String keyspace, Session session) {
IndexMetadata imd = session.getCluster().getMetadata().getKeyspace(keyspace.toLowerCase())
.getTable(expected.getTableName().toCql()).getColumn(expected.getColumnName().toCql()).getIndex();
TableMetadata tableMetadata = session.getCluster().getMetadata().getKeyspace(keyspace.toLowerCase())
.getTable(expected.getTableName().toCql());
IndexMetadata imd = tableMetadata.getIndex(expected.getName().toCql());
assertEquals(expected.getName(), imd.getName());
assertEquals(expected.getName(), imd == null ? null : imd.getName());
}
public static void assertNoIndex(IndexDescriptor expected, String keyspace, Session session) {
IndexMetadata imd = session.getCluster().getMetadata().getKeyspace(keyspace.toLowerCase())
.getTable(expected.getTableName().toCql()).getColumn(expected.getColumnName().toCql()).getIndex();
TableMetadata tableMetadata = session.getCluster().getMetadata().getKeyspace(keyspace.toLowerCase())
.getTable(expected.getTableName().toCql());
IndexMetadata imd = tableMetadata.getIndex(expected.getName().toCql());
assertNull(imd);
}

6
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/cql/generator/CqlTableSpecificationAssertions.java
View File

@@ -31,7 +31,7 @@ import org.springframework.cassandra.core.keyspace.TableOption;
import com.datastax.driver.core.ColumnMetadata;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.TableMetadata;
import com.datastax.driver.core.TableMetadata.Options;
import com.datastax.driver.core.TableOptionsMetadata;
/**
* @author Matthew T. Adams
@@ -70,7 +70,7 @@ public class CqlTableSpecificationAssertions {
assertColumns(expected.getPrimaryKeyColumns(), actual.getPrimaryKey());
}
public static void assertOptions(Map<String, Object> expected, Options actual) {
public static void assertOptions(Map<String, Object> expected, TableOptionsMetadata actual) {
for (String key : expected.keySet()) {
@@ -143,7 +143,7 @@ public class CqlTableSpecificationAssertions {
}
@SuppressWarnings("unchecked")
public static <T> T getOptionFor(TableOption option, Class<?> type, Options options) {
public static <T> T getOptionFor(TableOption option, Class<?> type, TableOptionsMetadata options) {
switch (option) {
case BLOOM_FILTER_FP_CHANCE:
return (T) (Double) options.getBloomFilterFalsePositiveChance();

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

@@ -108,6 +108,9 @@ partitioner: org.apache.cassandra.dht.Murmur3Partitioner
data_file_directories:
- target/embeddedCassandra/data
hints_directory:
- target/embeddedCassandra/hints
# commit log
commitlog_directory: target/embeddedCassandra/commitlog

2
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/convert/CassandraConverters.java
View File

@@ -89,7 +89,7 @@ abstract class CassandraConverters {
@Override
public Date convert(Row row) {
return row.getDate(0);
return row.getTimestamp(0);
}
}

7
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/convert/ColumnReader.java
View File

@@ -4,6 +4,7 @@ import java.util.List;
import org.springframework.cassandra.core.cql.CqlIdentifier;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.Row;
@@ -50,15 +51,15 @@ public class ColumnReader {
List<DataType> collectionTypes = type.getTypeArguments();
if (collectionTypes.size() == 2) {
return row.getMap(i, collectionTypes.get(0).asJavaClass(), collectionTypes.get(1).asJavaClass());
return row.getMap(i, CodecRegistry.DEFAULT_INSTANCE.codecFor(collectionTypes.get(0)).getJavaType().getRawType(), CodecRegistry.DEFAULT_INSTANCE.codecFor(collectionTypes.get(1)).getJavaType().getRawType());
}
if (type.equals(DataType.list(collectionTypes.get(0)))) {
return row.getList(i, collectionTypes.get(0).asJavaClass());
return row.getList(i, CodecRegistry.DEFAULT_INSTANCE.codecFor(collectionTypes.get(0)).getJavaType().getRawType());
}
if (type.equals(DataType.set(collectionTypes.get(0)))) {
return row.getSet(i, collectionTypes.get(0).asJavaClass());
return row.getSet(i, CodecRegistry.DEFAULT_INSTANCE.codecFor(collectionTypes.get(0)).getJavaType().getRawType());
}
throw new IllegalStateException("Unknown Collection type encountered. Valid collections are Set, List and Map.");

9
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/convert/MappingCassandraConverter.java
View File

@@ -46,6 +46,7 @@ import org.springframework.data.util.TypeInformation;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.querybuilder.Delete.Where;
import com.datastax.driver.core.querybuilder.Insert;
@@ -254,7 +255,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
public void doWithPersistentProperty(CassandraPersistentProperty prop) {
Object value = accessor.getProperty(prop,
prop.isCompositePrimaryKey() ? prop.getType() : prop.getDataType().asJavaClass());
prop.isCompositePrimaryKey() ? prop.getType() : CodecRegistry.DEFAULT_INSTANCE.codecFor(prop.getDataType()).getJavaType().getRawType());
if (log.isDebugEnabled()) {
log.debug("doWithProperties Property.type {}, Property.value {}", prop.getType().getName(), value);
@@ -293,7 +294,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
public void doWithPersistentProperty(CassandraPersistentProperty prop) {
Object value = accessor.getProperty(prop,
prop.isCompositePrimaryKey() ? prop.getType() : prop.getDataType().asJavaClass());
prop.isCompositePrimaryKey() ? prop.getType() : CodecRegistry.DEFAULT_INSTANCE.codecFor(prop.getDataType()).getJavaType().getRawType());
if (prop.isCompositePrimaryKey()) {
CassandraPersistentEntity<?> keyEntity = prop.getCompositePrimaryKeyEntity();
@@ -326,7 +327,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
@Override
public void doWithPersistentProperty(CassandraPersistentProperty prop) {
Object value = accessor.getProperty(prop, prop.getDataType().asJavaClass());
Object value = accessor.getProperty(prop, CodecRegistry.DEFAULT_INSTANCE.codecFor(prop.getDataType()).getJavaType().getRawType());
where.and(QueryBuilder.eq(prop.getColumnName().toCql(), value));
}
});
@@ -388,7 +389,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
CassandraPersistentProperty idProperty = entity.getIdProperty();
if (idProperty != null) {
return wrapper.getProperty(entity.getIdProperty(),
idProperty.isCompositePrimaryKey() ? idProperty.getType() : idProperty.getDataType().asJavaClass());
idProperty.isCompositePrimaryKey() ? idProperty.getType() : CodecRegistry.DEFAULT_INSTANCE.codecFor(idProperty.getDataType()).getJavaType().getRawType());
}
// if the class doesn't have an id property, then it's using MapId

3
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/convert/RowReaderPropertyAccessor.java
View File

@@ -21,6 +21,7 @@ import org.springframework.expression.EvaluationContext;
import org.springframework.expression.PropertyAccessor;
import org.springframework.expression.TypedValue;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.ProtocolVersion;
import com.datastax.driver.core.Row;
@@ -52,7 +53,7 @@ enum RowReaderPropertyAccessor implements PropertyAccessor {
}
DataType columnType = row.getColumnDefinitions().getType(name);
ByteBuffer bytes = row.getBytes(name);
Object object = columnType.deserialize(bytes, ProtocolVersion.NEWEST_SUPPORTED);
Object object = CodecRegistry.DEFAULT_INSTANCE.codecFor(columnType).deserialize(bytes, ProtocolVersion.NEWEST_SUPPORTED);
return new TypedValue(object);
}

32
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraPersistentProperty.java
View File

@@ -188,30 +188,22 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
private DataType getDataTypeFor(CassandraType annotation) {
DataType.Name type = annotation.type();
switch (type) {
if (type.isCollection()) {
switch (type) {
case MAP:
ensureTypeArguments(annotation.typeArguments().length, 2);
return DataType.map(getDataTypeFor(annotation.typeArguments()[0]),
getDataTypeFor(annotation.typeArguments()[1]));
case MAP:
ensureTypeArguments(annotation.typeArguments().length, 2);
return DataType.map(getDataTypeFor(annotation.typeArguments()[0]),
getDataTypeFor(annotation.typeArguments()[1]));
case LIST:
ensureTypeArguments(annotation.typeArguments().length, 1);
return DataType.list(getDataTypeFor(annotation.typeArguments()[0]));
case LIST:
ensureTypeArguments(annotation.typeArguments().length, 1);
return DataType.list(getDataTypeFor(annotation.typeArguments()[0]));
case SET:
ensureTypeArguments(annotation.typeArguments().length, 1);
return DataType.set(getDataTypeFor(annotation.typeArguments()[0]));
default:
throw new InvalidDataAccessApiUsageException(String.format(
"unknown multivalued DataType [%s] for property [%s] in entity [%s]", type, getType(), getOwner()
.getName()));
}
} else {
case SET:
ensureTypeArguments(annotation.typeArguments().length, 1);
return DataType.set(getDataTypeFor(annotation.typeArguments()[0]));
default:
return CassandraSimpleTypeHolder.getDataTypeFor(type);
}
}

7
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/CassandraSimpleTypeHolder.java
View File

@@ -26,6 +26,7 @@ import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.mapping.model.SimpleTypeHolder;
import org.springframework.data.util.TypeInformation;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.DataType.Name;
@@ -87,9 +88,11 @@ public class CassandraSimpleTypeHolder extends SimpleTypeHolder {
Map<Class<?>, DataType> classToDataType = new HashMap<Class<?>, DataType>(16);
CodecRegistry codecRegistry = CodecRegistry.DEFAULT_INSTANCE;
for (DataType dataType : DataType.allPrimitiveTypes()) {
Class<?> javaClass = dataType.asJavaClass();
Class<?> javaClass = codecRegistry.codecFor(dataType).getJavaType().getRawType();
classToDataType.put(javaClass, dataType);
Class<?> primitiveJavaClass = primitiveWrappers.get(javaClass);
@@ -114,7 +117,7 @@ public class CassandraSimpleTypeHolder extends SimpleTypeHolder {
Set<Class<?>> simpleTypes = new HashSet<Class<?>>();
for (DataType dataType : DataType.allPrimitiveTypes()) {
Class<?> javaClass = dataType.asJavaClass();
Class<?> javaClass = CodecRegistry.DEFAULT_INSTANCE.codecFor(dataType).getJavaType().getRawType();
simpleTypes.add(javaClass);
}
return simpleTypes;

2
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/convert/MappingCassandraConverterUnitTests.java
View File

@@ -389,7 +389,7 @@ public class MappingCassandraConverterUnitTests {
public void shouldReadDateCorrectly() throws UnknownHostException {
Date date = new Date(1);
when(rowMock.getDate(0)).thenReturn(date);
when(rowMock.getTimestamp(0)).thenReturn(date);
Date result = mappingCassandraConverter.readRow(Date.class, rowMock);