Add support for Cassandra

Add connectors for Cassandra Cluster instances with Java and XML-based configuration. Cassandra configuration is obtained using the single-tenant Datastax/Cassandra tile.

build.gradle

@@ -34,6 +34,9 @@ ext {
 	mariadbDriverVersion = "1.1.3"
 	postgresDriverVersion = "9.0-801.jdbc4"
 
+	springDataCassandraVersion = "1.5.3.RELEASE"
+	cassandraDriverVersion = "3.1.3"
+
 	javaxMailVersion = "1.4.7"
 
 	cglibVersion = "3.1"
@@ -66,7 +69,7 @@ subprojects {
 	apply plugin: 'propdeps-eclipse'
 	apply plugin: "org.asciidoctor.gradle.asciidoctor"
 
-	asciidoctor { 
+	asciidoctor {
 		sourceDir = new File("docs/src/main/asciidoc")
 		outputDir = new File("docs/target/generated-docs")
 		options = [
@@ -208,6 +211,9 @@ ext {
 			"jedis26-redis15"  : [jedisVersion: "2.6.3", springDataRedisVersion: "1.5.2.RELEASE"],
 			"jedis27-redis16"  : [jedisVersion: "2.7.3", springDataRedisVersion: "1.6.4.RELEASE"],
 			"jedis28-redis17"  : [jedisVersion: "2.8.1", springDataRedisVersion: "1.7.2.RELEASE"],
+			"driver300-cass15" : [cassandraDriverVersion: "3.0.0", springDataCassandraVersion: "1.5.3.RELEASE"],
+			"driver301-cass15" : [cassandraDriverVersion: "3.0.1", springDataCassandraVersion: "1.5.3.RELEASE"],
+			"driver313-cass15" : [cassandraDriverVersion: "3.1.3", springDataCassandraVersion: "1.5.3.RELEASE"],
 //			"lettuce34-redis15": [lettuceVersion: "3.4.3.Final", springDataRedisVersion: "1.5.2.RELEASE"],
 //			"lettuce34-redis16": [lettuceVersion: "3.4.3.Final", springDataRedisVersion: "1.6.4.RELEASE"],
 //			"lettuce34-redis17": [lettuceVersion: "3.4.3.Final", springDataRedisVersion: "1.7.2.RELEASE"],

spring-cloud-cloudfoundry-connector/src/main/java/org/springframework/cloud/cloudfoundry/CassandraServiceInfoCreator.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.cloudfoundry;
+
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.cloud.service.common.CassandraServiceInfo;
+
+
+/**
+ * Service info creator for Cassandra services.
+ *
+ * @author Mark Paluch
+ */
+public class CassandraServiceInfoCreator extends
+		CloudFoundryServiceInfoCreator<CassandraServiceInfo> {
+
+	public CassandraServiceInfoCreator() {
+		super(new Tags("cassandra"));
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public CassandraServiceInfo createServiceInfo(Map<String, Object> serviceData) {
+
+		String id = (String) serviceData.get("name");
+
+		Map<String, Object> credentials = getCredentials(serviceData);
+
+		String username = getStringFromCredentials(credentials, "username");
+		String password = getStringFromCredentials(credentials, "password");
+		String port = getStringFromCredentials(credentials, "cqlsh_port");
+		List<String> contactpoints = (List<String>) credentials.get("node_ips");
+
+		return new CassandraServiceInfo(id, contactpoints, Integer.parseInt(port),
+				username, password);
+	}
+}

spring-cloud-cloudfoundry-connector/src/main/resources/META-INF/services/org.springframework.cloud.cloudfoundry.CloudFoundryServiceInfoCreator

@@ -8,3 +8,4 @@ org.springframework.cloud.cloudfoundry.SmtpServiceInfoCreator
 org.springframework.cloud.cloudfoundry.OracleServiceInfoCreator
 org.springframework.cloud.cloudfoundry.DB2ServiceInfoCreator
 org.springframework.cloud.cloudfoundry.SqlServerServiceInfoCreator
+org.springframework.cloud.cloudfoundry.CassandraServiceInfoCreator

spring-cloud-cloudfoundry-connector/src/test/java/org/springframework/cloud/cloudfoundry/CassandraServiceInfoCreatorTests.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.cloudfoundry;
+
+import static org.hamcrest.CoreMatchers.*;
+import static org.junit.Assert.*;
+
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Test;
+import org.springframework.cloud.service.common.CassandraServiceInfo;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * Unit tests for link {@link CassandraServiceInfoCreator}.
+ *
+ * @author Mark Paluch
+ */
+public class CassandraServiceInfoCreatorTests extends AbstractCloudFoundryConnectorTest {
+
+	private ObjectMapper mapper = new ObjectMapper();
+
+	@Test
+	public void shouldCreateServiceInfo() throws Exception {
+
+		CassandraServiceInfoCreator creator = new CassandraServiceInfoCreator();
+		Map services = readServiceData("test-cassandra-service.json");
+		Map<String, Object> serviceData = getServiceData(services,
+				"p-dse-cassandra-acceptance");
+
+		CassandraServiceInfo info = creator.createServiceInfo(serviceData);
+
+		assertThat(info.getContactPoints(), hasItems("1.2.3.4", "5.6.7.8"));
+		assertThat(info.getPort(), is(equalTo(9042)));
+		assertThat(info.getUsername(), is(nullValue()));
+		assertThat(info.getPassword(), is(nullValue()));
+	}
+
+	@Test
+	public void shouldCreateServiceInfoWithCredentials() throws Exception {
+
+		CassandraServiceInfoCreator creator = new CassandraServiceInfoCreator();
+		Map services = readServiceData("test-cassandra-with-credentials.json");
+		Map<String, Object> serviceData = getServiceData(services,
+				"p-dse-cassandra-acceptance");
+
+		CassandraServiceInfo info = creator.createServiceInfo(serviceData);
+
+		assertThat(info.getContactPoints(), hasItems("1.2.3.4"));
+		assertThat(info.getPort(), is(equalTo(9042)));
+		assertThat(info.getUsername(), is(equalTo("user")));
+		assertThat(info.getPassword(), is(equalTo("pass")));
+	}
+
+	@Test
+	public void testAcceptService() throws Exception {
+
+		CassandraServiceInfoCreator creator = new CassandraServiceInfoCreator();
+		Map services = readServiceData("test-cassandra-service.json");
+		Map<String, Object> serviceData = getServiceData(services,
+				"p-dse-cassandra-acceptance");
+
+		assertThat(creator.accept(serviceData), is(true));
+	}
+
+	private Map readServiceData(String resource) throws java.io.IOException {
+		return mapper.readValue(readTestDataFile(resource), Map.class);
+	}
+
+	@SuppressWarnings("unchecked")
+	private Map<String, Object> getServiceData(Map services, String name) {
+		return (Map<String, Object>) ((List) services.get(name)).get(0);
+	}
+}

spring-cloud-cloudfoundry-connector/src/test/resources/org/springframework/cloud/cloudfoundry/test-cassandra-service.json

@@ -0,0 +1,25 @@
+{
+  "p-dse-cassandra-acceptance": [
+	{
+	  "credentials": {
+		"cqlsh_port": "9042",
+		"node_ips": [
+		  "1.2.3.4",
+		  "5.6.7.8"
+		],
+		"opscenter_url": "http://datastax-opscenter-cassandra.cf-app.com/opscenter/index.html"
+	  },
+	  "label": "p-dse-cassandra-acceptance",
+	  "name": "mydse",
+	  "plan": "single-node",
+	  "provider": null,
+	  "syslog_drain_url": null,
+	  "tags": [
+		"pivotal",
+		"cassandra",
+		"dse",
+		"datastax"
+	  ]
+	}
+  ]
+}

spring-cloud-cloudfoundry-connector/src/test/resources/org/springframework/cloud/cloudfoundry/test-cassandra-with-credentials.json

@@ -0,0 +1,26 @@
+{
+  "p-dse-cassandra-acceptance": [
+	{
+	  "credentials": {
+		"cqlsh_port": "9042",
+		"node_ips": [
+		  "1.2.3.4"
+		],
+		"opscenter_url": "http://datastax-opscenter-cassandra.cf-app.com/opscenter/index.html",
+		"password": "pass",
+		"username": "user"
+	  },
+	  "label": "p-dse-cassandra-acceptance",
+	  "name": "mydse",
+	  "plan": "single-node",
+	  "provider": null,
+	  "syslog_drain_url": null,
+	  "tags": [
+		"pivotal",
+		"cassandra",
+		"dse",
+		"datastax"
+	  ]
+	}
+  ]
+}

spring-cloud-core/src/main/java/org/springframework/cloud/service/common/CassandraServiceInfo.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.service.common;
+
+import java.util.List;
+
+import org.springframework.cloud.service.BaseServiceInfo;
+import org.springframework.cloud.service.ServiceInfo.ServiceLabel;
+
+/**
+ * {@link org.springframework.cloud.service.ServiceInfo} for a Cassandra/Datastax DSE
+ * service.
+ *
+ * @author Mark Paluch
+ */
+@ServiceLabel("cassandra")
+public class CassandraServiceInfo extends BaseServiceInfo {
+
+	private final List<String> contactPoints;
+	private final int port;
+	private final String username;
+	private final String password;
+
+	/**
+	 * Creates a new {@link CassandraServiceInfo} using Contact points and port.
+	 *
+	 * @param id the service-id
+	 * @param contactPoints list of contact-points
+	 * @param port the port
+	 */
+	public CassandraServiceInfo(String id, List<String> contactPoints, int port) {
+		this(id, contactPoints, port, null, null);
+	}
+
+	/**
+	 * Creates a new {@link CassandraServiceInfo} using Contact points with a port and
+	 * username/password credentials.
+	 *
+	 * @param id the service-id
+	 * @param contactPoints list of contact-points
+	 * @param port the port
+	 * @param username the user name
+	 * @param password the password
+	 */
+	public CassandraServiceInfo(String id, List<String> contactPoints, int port,
+			String username, String password) {
+		super(id);
+
+		this.contactPoints = contactPoints;
+		this.port = port;
+		this.username = username;
+		this.password = password;
+	}
+
+	@ServiceProperty(category = "contactpoints")
+	public List<String> getContactPoints() {
+		return contactPoints;
+	}
+
+	@ServiceProperty(category = "port")
+	public int getPort() {
+		return port;
+	}
+
+	@ServiceProperty(category = "username")
+	public String getUsername() {
+		return username;
+	}
+
+	@ServiceProperty(category = "password")
+	public String getPassword() {
+		return password;
+	}
+}

spring-cloud-core/src/test/java/org/springframework/cloud/service/common/CassandraServiceInfoUnitTests.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.service.common;
+
+import static org.hamcrest.CoreMatchers.*;
+import static org.junit.Assert.assertThat;
+
+import java.util.Collections;
+
+
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link CassandraServiceInfo}.
+ *
+ * @author Mark Paluch
+ */
+public class CassandraServiceInfoUnitTests {
+
+	@Test
+	public void shouldContainContactPointsAndPort() {
+
+		CassandraServiceInfo info = new CassandraServiceInfo("cassandra",
+				Collections.singletonList("10.0.0.1"), 9042);
+
+		assertThat(info.getContactPoints(), hasItem("10.0.0.1"));
+		assertThat(info.getPort(), is(equalTo(9042)));
+	}
+
+	@Test
+	public void shouldContainContactPointsAndPortAndCredentials() {
+
+		CassandraServiceInfo info = new CassandraServiceInfo("cassandra",
+				Collections.singletonList("10.0.0.1"), 9042, "walter", "white");
+
+		assertThat(info.getContactPoints(), hasItem("10.0.0.1"));
+		assertThat(info.getPort(), is(equalTo(9042)));
+		assertThat(info.getUsername(), is(equalTo("walter")));
+		assertThat(info.getPassword(), is(equalTo("white")));
+	}
+}

spring-cloud-spring-service-connector/build.gradle

@@ -40,4 +40,14 @@ dependencies {
     exclude(group: 'org.mongodb', module: 'mongo-java-driver')
   }
   optional("org.mongodb:mongo-java-driver:${mongoDriverVersion}")
+
+
+  optional("org.springframework.data:spring-data-cassandra:$springDataCassandraVersion") {
+    exclude(group: 'org.springframework', module: 'spring-beans')
+    exclude(group: 'org.springframework', module: 'spring-expression')
+    exclude(group: 'org.springframework', module: 'spring-tx')
+    // depend on cassandra-driver-core explicitly to control version used for testing
+    exclude(group: 'com.datastax.cassandra', module: 'cassandra-driver-core')
+  }
+  optional("com.datastax.cassandra:cassandra-driver-core:${cassandraDriverVersion}")
 }

spring-cloud-spring-service-connector/src/main/java/org/springframework/cloud/config/java/CloudServiceConnectionFactory.java

@@ -1,10 +1,12 @@
 package org.springframework.cloud.config.java;
 
+import com.datastax.driver.core.Cluster;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.cloud.Cloud;
 import org.springframework.cloud.CloudException;
 import org.springframework.cloud.service.PooledServiceConnectorConfig;
 import org.springframework.cloud.service.ServiceConnectorConfig;
+import org.springframework.cloud.service.column.CassandraClusterConfig;
 import org.springframework.cloud.service.document.MongoDbFactoryConfig;
 import org.springframework.cloud.service.messaging.RabbitConnectionFactoryConfig;
 import org.springframework.cloud.service.relational.DataSourceConfig;
@@ -294,6 +296,62 @@ public class CloudServiceConnectionFactory implements ServiceConnectionFactory {
 		return cloud.getServiceConnector(serviceId, RedisConnectionFactory.class, redisConnectionFactoryConfig);
 	}
 
+	/**
+	 * Get the {@link Cluster} object associated with the only Cassandra service bound to
+	 * the app.
+	 *
+	 * @return the Cassandra {@link Cluster}
+	 * @throws org.springframework.cloud.CloudException if there are either 0 or more than
+	 * 1 Cassandra database services.
+	 */
+	@Override
+	public Cluster cluster() {
+		return cluster((CassandraClusterConfig) null);
+	}
+
+	/**
+	 * Get the {@link Cluster} object associated with the only Cassandra service bound to
+	 * the app configured as specified.
+	 *
+	 * @param config configuration for the Cluster created
+	 * @return the Cassandra {@link Cluster}
+	 * @throws org.springframework.cloud.CloudException if there are either 0 or more than
+	 * 1 Cassandra database services.
+	 */
+	@Override
+	public Cluster cluster(CassandraClusterConfig config) {
+		return cloud.getSingletonServiceConnector(Cluster.class, config);
+	}
+
+	/**
+	 * Get the {@link Cluster} object associated with the Cassandra {@code serviceId}
+	 * bound to the app.
+	 *
+	 * @param serviceId the Cassandra serviceId
+	 * @return the Cassandra {@link Cluster}
+	 * @throws org.springframework.cloud.CloudException if there is no service bound with
+	 * the {@code serviceId}
+	 */
+	@Override
+	public Cluster cluster(String serviceId) {
+		return cluster(serviceId, null);
+	}
+
+	/**
+	 * Get the {@link Cluster} object associated with the Cassandra {@code serviceId}
+	 * bound to the app configured as specified.
+	 *
+	 * @param serviceId the Cassandra serviceId
+	 * @param config configuration for the Cluster created
+	 * @return the Cassandra {@link Cluster}
+	 * @throws org.springframework.cloud.CloudException if there is no service bound with
+	 * the {@code serviceId}
+	 */
+	@Override
+	public Cluster cluster(String serviceId, CassandraClusterConfig config) {
+		return cloud.getServiceConnector(serviceId, Cluster.class, config);
+	}
+
 	// Generic service
 	/**
 	 * Get the service connector object associated with the only service bound to the app.

spring-cloud-spring-service-connector/src/main/java/org/springframework/cloud/config/java/ServiceConnectionFactory.java

@@ -1,8 +1,10 @@
 package org.springframework.cloud.config.java;
 
+import com.datastax.driver.core.Cluster;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.cloud.service.PooledServiceConnectorConfig;
 import org.springframework.cloud.service.ServiceConnectorConfig;
+import org.springframework.cloud.service.column.CassandraClusterConfig;
 import org.springframework.cloud.service.document.MongoDbFactoryConfig;
 import org.springframework.cloud.service.messaging.RabbitConnectionFactoryConfig;
 import org.springframework.cloud.service.relational.DataSourceConfig;
@@ -49,6 +51,50 @@ public interface ServiceConnectionFactory {
 	RedisConnectionFactory redisConnectionFactory(String serviceId,
 												  PooledServiceConnectorConfig redisConnectionFactoryConfig);
 
+	/**
+	 * Get the {@link Cluster} object associated with the only Cassandra service bound to
+	 * the app.
+	 *
+	 * @return the Cassandra {@link Cluster}
+	 * @throws org.springframework.cloud.CloudException if there are either 0 or more than
+	 * 1 Cassandra database services.
+	 */
+	Cluster cluster();
+
+	/**
+	 * Get the {@link Cluster} object associated with the only Cassandra service bound to
+	 * the app configured as specified.
+	 *
+	 * @param config configuration for the Cluster created
+	 * @return the Cassandra {@link Cluster}
+	 * @throws org.springframework.cloud.CloudException if there are either 0 or more than
+	 * 1 Cassandra database services.
+	 */
+	Cluster cluster(CassandraClusterConfig config);
+
+	/**
+	 * Get the {@link Cluster} object associated with the Cassandra {@code serviceId}
+	 * bound to the app.
+	 *
+	 * @param serviceId the Cassandra serviceId
+	 * @return the Cassandra {@link Cluster}
+	 * @throws org.springframework.cloud.CloudException if there is no service bound with
+	 * the {@code serviceId}
+	 */
+	Cluster cluster(String serviceId);
+
+	/**
+	 * Get the {@link Cluster} object associated with the Cassandra {@code serviceId}
+	 * bound to the app configured as specified.
+	 *
+	 * @param serviceId the Cassandra serviceId
+	 * @param config configuration for the Cluster created
+	 * @return the Cassandra {@link Cluster}
+	 * @throws org.springframework.cloud.CloudException if there is no service bound with
+	 * the {@code serviceId}
+	 */
+	Cluster cluster(String serviceId, CassandraClusterConfig config);
+
 	Object service();
 
 	<T> T service(Class<T> serviceConnectorType);

spring-cloud-spring-service-connector/src/main/java/org/springframework/cloud/config/xml/CloudCassandraSessionParser.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.config.xml;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.cloud.service.column.CassandraClusterConfig;
+import org.springframework.cloud.service.column.CassandraClusterFactory;
+import org.springframework.util.StringUtils;
+import org.springframework.util.xml.DomUtils;
+import org.w3c.dom.Element;
+
+import com.datastax.driver.core.ProtocolOptions;
+
+/**
+ * @author Vinicius Carvalho
+ */
+public class CloudCassandraSessionParser
+		extends AbstractNestedElementCloudServiceFactoryParser {
+
+	final String ELEMENT_CASSANDRA_OPTIONS = "cassandra-options";
+	final String COMPRESSION_ATTRIBUTE = "compression";
+	final String RETRY_POLICY_ATTRIBUTE = "retry-policy";
+	final String LOAD_BALANCING_POLICY_ATTRIBUTE = "loadbalancing-policy";
+	final String SOCKET_OPTIONS_ATTRIBUTE = "socket-options";
+	final String RECONNECTION_POLICY_ATTRIBUTE = "reconnection-policy";
+
+	public CloudCassandraSessionParser() {
+		super(CassandraClusterFactory.class);
+	}
+
+	@Override
+	protected void doParse(Element element, ParserContext parserContext,
+			BeanDefinitionBuilder builder) {
+		super.doParse(element, parserContext, builder);
+
+		Element optionsElement = DomUtils.getChildElementByTagName(element,
+				ELEMENT_CASSANDRA_OPTIONS);
+
+		BeanDefinitionBuilder beanBuilder = BeanDefinitionBuilder
+				.genericBeanDefinition(CassandraClusterConfig.class.getName());
+
+		if (optionsElement != null) {
+			String compressionString = optionsElement.getAttribute(COMPRESSION_ATTRIBUTE);
+			if (!StringUtils.isEmpty(compressionString)) {
+				ProtocolOptions.Compression compression = ProtocolOptions.Compression
+						.valueOf(compressionString);
+				beanBuilder.addPropertyValue("compression", compression);
+			}
+
+			String retryPolicyString = optionsElement
+					.getAttribute(RETRY_POLICY_ATTRIBUTE);
+			if (!StringUtils.isEmpty(retryPolicyString)) {
+				beanBuilder.addPropertyReference("retryPolicy", retryPolicyString);
+			}
+
+			String loadBalancingPolicyString = optionsElement
+					.getAttribute(LOAD_BALANCING_POLICY_ATTRIBUTE);
+			if (!StringUtils.isEmpty(loadBalancingPolicyString)) {
+				beanBuilder.addPropertyReference("loadBalancingPolicy",
+						loadBalancingPolicyString);
+			}
+
+			String socketOptionsString = optionsElement
+					.getAttribute(SOCKET_OPTIONS_ATTRIBUTE);
+			if (!StringUtils.isEmpty(socketOptionsString)) {
+				beanBuilder.addPropertyReference("socketOptions", socketOptionsString);
+			}
+
+			String reconnectionPolicyString = optionsElement
+					.getAttribute(RECONNECTION_POLICY_ATTRIBUTE);
+			if (!StringUtils.isEmpty(reconnectionPolicyString)) {
+				beanBuilder.addPropertyReference("reconnectionPolicy",
+						reconnectionPolicyString);
+			}
+
+		}
+
+		builder.addConstructorArgValue(beanBuilder.getBeanDefinition());
+	}
+
+}

spring-cloud-spring-service-connector/src/main/java/org/springframework/cloud/config/xml/CloudNamespaceHandler.java

@@ -25,6 +25,7 @@ public class CloudNamespaceHandler extends NamespaceHandlerSupport {
 		registerBeanDefinitionParser("redis-connection-factory", new CloudRedisConnectionFactoryParser());
 		registerBeanDefinitionParser("mongo-db-factory", new CloudMongoDbFactoryParser());
 		registerBeanDefinitionParser("data-source", new CloudDataSourceFactoryParser());
+		registerBeanDefinitionParser("cassandra-session-factory", new CloudCassandraSessionParser());
 
 		registerBeanDefinitionParser("connection-properties", new ConnectionPropertiesParser());
 

spring-cloud-spring-service-connector/src/main/java/org/springframework/cloud/service/column/CassandraClusterConfig.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.service.column;
+
+import org.springframework.cloud.service.ServiceConnectorConfig;
+
+import com.datastax.driver.core.NettyOptions;
+import com.datastax.driver.core.PoolingOptions;
+import com.datastax.driver.core.ProtocolOptions.Compression;
+import com.datastax.driver.core.ProtocolVersion;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.policies.LoadBalancingPolicy;
+import com.datastax.driver.core.policies.ReconnectionPolicy;
+import com.datastax.driver.core.policies.RetryPolicy;
+
+/**
+ * Configuration for a Cassandra {@link com.datastax.driver.core.Cluster}.
+ *
+ * @author Mark Paluch
+ */
+public class CassandraClusterConfig implements ServiceConnectorConfig {
+
+	public static final boolean DEFAULT_METRICS_ENABLED = true;
+	public static final boolean DEFAULT_JMX_REPORTING_ENABLED = true;
+
+	// Protocol options
+	private Compression compression;
+	private NettyOptions nettyOptions;
+	private ProtocolVersion protocolVersion;
+
+	// Policies
+	private LoadBalancingPolicy loadBalancingPolicy;
+	private ReconnectionPolicy reconnectionPolicy;
+	private RetryPolicy retryPolicy;
+
+	private PoolingOptions poolingOptions;
+	private QueryOptions queryOptions;
+	private SocketOptions socketOptions;
+
+	private boolean metricsEnabled = DEFAULT_METRICS_ENABLED;
+	private boolean jmxReportingEnabled = DEFAULT_JMX_REPORTING_ENABLED;
+
+	public Compression getCompression() {
+		return compression;
+	}
+
+	public void setCompression(Compression compression) {
+		this.compression = compression;
+	}
+
+	public NettyOptions getNettyOptions() {
+		return nettyOptions;
+	}
+
+	public void setNettyOptions(NettyOptions nettyOptions) {
+		this.nettyOptions = nettyOptions;
+	}
+
+	public ProtocolVersion getProtocolVersion() {
+		return protocolVersion;
+	}
+
+	public void setProtocolVersion(ProtocolVersion protocolVersion) {
+		this.protocolVersion = protocolVersion;
+	}
+
+	public LoadBalancingPolicy getLoadBalancingPolicy() {
+		return loadBalancingPolicy;
+	}
+
+	public void setLoadBalancingPolicy(LoadBalancingPolicy loadBalancingPolicy) {
+		this.loadBalancingPolicy = loadBalancingPolicy;
+	}
+
+	public ReconnectionPolicy getReconnectionPolicy() {
+		return reconnectionPolicy;
+	}
+
+	public void setReconnectionPolicy(ReconnectionPolicy reconnectionPolicy) {
+		this.reconnectionPolicy = reconnectionPolicy;
+	}
+
+	public RetryPolicy getRetryPolicy() {
+		return retryPolicy;
+	}
+
+	public void setRetryPolicy(RetryPolicy retryPolicy) {
+		this.retryPolicy = retryPolicy;
+	}
+
+	public PoolingOptions getPoolingOptions() {
+		return poolingOptions;
+	}
+
+	public void setPoolingOptions(PoolingOptions poolingOptions) {
+		this.poolingOptions = poolingOptions;
+	}
+
+	public QueryOptions getQueryOptions() {
+		return queryOptions;
+	}
+
+	public void setQueryOptions(QueryOptions queryOptions) {
+		this.queryOptions = queryOptions;
+	}
+
+	public SocketOptions getSocketOptions() {
+		return socketOptions;
+	}
+
+	public void setSocketOptions(SocketOptions socketOptions) {
+		this.socketOptions = socketOptions;
+	}
+
+	public boolean isMetricsEnabled() {
+		return metricsEnabled;
+	}
+
+	public void setMetricsEnabled(boolean metricsEnabled) {
+		this.metricsEnabled = metricsEnabled;
+	}
+
+	public boolean isJmxReportingEnabled() {
+		return jmxReportingEnabled;
+	}
+
+	public void setJmxReportingEnabled(boolean jmxReportingEnabled) {
+		this.jmxReportingEnabled = jmxReportingEnabled;
+	}
+}

spring-cloud-spring-service-connector/src/main/java/org/springframework/cloud/service/column/CassandraClusterCreator.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.service.column;
+
+import org.springframework.cloud.service.AbstractServiceConnectorCreator;
+import org.springframework.cloud.service.ServiceConnectorConfig;
+import org.springframework.cloud.service.ServiceConnectorCreator;
+import org.springframework.cloud.service.common.CassandraServiceInfo;
+import org.springframework.util.StringUtils;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Cluster.Builder;
+
+/**
+ * {@link ServiceConnectorCreator} implementation to provide a Cassandra {@link Cluster}.
+ * Allows optionally to apply a {@link CassandraClusterConfig} configuration.
+ *
+ * @author Mark Paluch
+ */
+public class CassandraClusterCreator
+		extends AbstractServiceConnectorCreator<Cluster, CassandraServiceInfo> {
+
+	@Override
+	public Cluster create(CassandraServiceInfo serviceInfo,
+			ServiceConnectorConfig serviceConnectorConfig) {
+
+		Builder builder = Cluster.builder()
+				.addContactPoints(serviceInfo.getContactPoints().toArray(new String[0]))
+				.withPort(serviceInfo.getPort());
+
+		if (StringUtils.hasText(serviceInfo.getUsername())) {
+			builder.withCredentials(serviceInfo.getUsername(), serviceInfo.getPassword());
+		}
+
+		if (serviceConnectorConfig instanceof CassandraClusterConfig) {
+
+			CassandraClusterConfig config = (CassandraClusterConfig) serviceConnectorConfig;
+
+			if (config.getCompression() != null) {
+				builder.withCompression(config.getCompression());
+			}
+
+			builder.withPoolingOptions(config.getPoolingOptions());
+			builder.withSocketOptions(config.getSocketOptions());
+			builder.withQueryOptions(config.getQueryOptions());
+			builder.withNettyOptions(config.getNettyOptions());
+			builder.withLoadBalancingPolicy(config.getLoadBalancingPolicy());
+			builder.withReconnectionPolicy(config.getReconnectionPolicy());
+			builder.withRetryPolicy(config.getRetryPolicy());
+			builder.withProtocolVersion(config.getProtocolVersion());
+
+			if (!config.isMetricsEnabled()) {
+				builder.withoutMetrics();
+			}
+
+			if (!config.isJmxReportingEnabled()) {
+				builder.withoutJMXReporting();
+			}
+		}
+
+		return builder.build();
+	}
+}

spring-cloud-spring-service-connector/src/main/java/org/springframework/cloud/service/column/CassandraClusterFactory.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.service.column;
+
+import org.springframework.cloud.service.AbstractCloudServiceConnectorFactory;
+import org.springframework.cloud.service.ServiceConnectorConfig;
+
+import com.datastax.driver.core.Cluster;
+
+/**
+ * Spring factory bean for Cassandra service.
+ *
+ * @author Mark Paluch
+ */
+public class CassandraClusterFactory
+		extends AbstractCloudServiceConnectorFactory<Cluster> {
+
+	public CassandraClusterFactory(String serviceId,
+			ServiceConnectorConfig serviceConnectorConfiguration) {
+		super(serviceId, Cluster.class, serviceConnectorConfiguration);
+	}
+}

spring-cloud-spring-service-connector/src/main/resources/META-INF/services/org.springframework.cloud.service.ServiceConnectorCreator

@@ -7,3 +7,4 @@ org.springframework.cloud.service.document.MongoDbFactoryCreator
 org.springframework.cloud.service.messaging.RabbitConnectionFactoryCreator
 org.springframework.cloud.service.smtp.MailSenderCreator
 org.springframework.cloud.service.relational.SqlServerDataSourceCreator
+org.springframework.cloud.service.column.CassandraClusterCreator

spring-cloud-spring-service-connector/src/main/resources/org/springframework/cloud/config/xml/spring-cloud.xsd

@@ -349,4 +349,115 @@
 		</xsd:complexContent>
 	</xsd:complexType>
 
+ 	<xsd:element name="cassandra-session-factory">
+ 		<xsd:annotation>
+ 			<xsd:documentation>
+ 				Locates a cassandra service
+ 			</xsd:documentation>
+ 		</xsd:annotation>
+ 
+ 		<xsd:complexType>
+ 
+ 			<xsd:sequence>
+ 				<xsd:element name="cassandra-options" type="cassandraOptionsType" minOccurs="0" maxOccurs="1"/>
+ 
+ 			</xsd:sequence>
+ 			<xsd:attribute name="id" type="xsd:string" use="optional">
+ 				<xsd:annotation>
+ 					<xsd:documentation>
+ 						The id for this Cassandra Session instance.
+ 						If not provided, the service name will be used as a fallback.
+ 					</xsd:documentation>
+ 				</xsd:annotation>
+ 			</xsd:attribute>
+ 			<xsd:attribute name="service-name" type="xsd:string" use="optional">
+ 				<xsd:annotation>
+ 					<xsd:documentation>
+ 						The name of the cassandra service.
+ 					</xsd:documentation>
+ 				</xsd:annotation>
+ 			</xsd:attribute>
+ 
+ 		</xsd:complexType>
+ 
+ 	</xsd:element>
+ 
+ 	<xsd:complexType name="cassandraOptionsType">
+ 		<xsd:annotation>
+ 			<xsd:documentation><![CDATA[
+ 				Element defining optional Cassandra configuration settings.
+ 			]]></xsd:documentation>
+ 		</xsd:annotation>
+ 		<xsd:attributeGroup ref="compressionOptions"/>
+ 		<xsd:attribute type="xsd:string" name="reconnection-policy">
+ 			<xsd:annotation>
+ 				<xsd:documentation><![CDATA[
+ 					A reference to a com.datastax.driver.core.policies.ReconnectionPolicy implementation
+ 	          		]]></xsd:documentation>
+ 				<xsd:appinfo>
+ 					<tool:annotation kind="ref">
+ 						<tool:expected-type type="com.datastax.driver.core.policies.ReconnectionPolicy"/>
+ 					</tool:annotation>
+ 				</xsd:appinfo>
+ 			</xsd:annotation>
+ 
+ 		</xsd:attribute>
+ 		<xsd:attribute type="xsd:string" name="loadbalancing-policy">
+ 			<xsd:annotation>
+ 				<xsd:documentation><![CDATA[
+ 					A reference to a com.datastax.driver.core.policies.LoadBalancingPolicy implementation
+ 	          		]]></xsd:documentation>
+ 				<xsd:appinfo>
+ 					<tool:annotation kind="ref">
+ 						<tool:expected-type type="com.datastax.driver.core.policies.LoadBalancingPolicy"/>
+ 					</tool:annotation>
+ 				</xsd:appinfo>
+ 			</xsd:annotation>
+ 
+ 		</xsd:attribute>
+ 		<xsd:attribute type="xsd:string" name="socket-options">
+ 			<xsd:annotation>
+ 				<xsd:documentation><![CDATA[
+ 					A reference to a com.datastax.driver.core.SocketOptions Object
+ 	          		]]></xsd:documentation>
+ 				<xsd:appinfo>
+ 					<tool:annotation kind="ref">
+ 						<tool:expected-type type="com.datastax.driver.core.SocketOptions"/>
+ 					</tool:annotation>
+ 				</xsd:appinfo>
+ 			</xsd:annotation>
+ 
+ 		</xsd:attribute>
+ 		<xsd:attribute type="xsd:string" name="retry-policy">
+ 			<xsd:annotation>
+ 				<xsd:documentation><![CDATA[
+ 					A reference to a com.datastax.driver.core.policies.RetryPolicy Object
+ 	          		]]></xsd:documentation>
+ 				<xsd:appinfo>
+ 					<tool:annotation kind="ref">
+ 						<tool:expected-type type="com.datastax.driver.core.policies.RetryPolicy"/>
+ 					</tool:annotation>
+ 				</xsd:appinfo>
+ 			</xsd:annotation>
+ 
+ 		</xsd:attribute>
+ 	</xsd:complexType>
+ 
+ 	<xsd:attributeGroup name="compressionOptions">
+ 		<xsd:attribute name="compression">
+ 			<xsd:annotation>
+ 				<xsd:documentation><![CDATA[
+ 					com.datastax.driver.core.policies.RetryPolicy to be used
+ 				]]></xsd:documentation>
+ 			</xsd:annotation>
+ 			<xsd:simpleType>
+ 				<xsd:restriction base="xsd:string">
+ 					<xsd:enumeration value="NONE" />
+ 					<xsd:enumeration value="LZ4" />
+ 					<xsd:enumeration value="SNAPPY" />
+ 				</xsd:restriction>
+ 			</xsd:simpleType>
+ 		</xsd:attribute>
+ 	</xsd:attributeGroup>
+ 
 </xsd:schema>

spring-cloud-spring-service-connector/src/test/java/org/springframework/cloud/StubCloudConnectorTest.java

@@ -1,7 +1,10 @@
 package org.springframework.cloud;
 
+import java.util.Arrays;
+
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.cloud.service.ServiceInfo;
+import org.springframework.cloud.service.common.CassandraServiceInfo;
 import org.springframework.cloud.service.common.MongoServiceInfo;
 import org.springframework.cloud.service.common.MysqlServiceInfo;
 import org.springframework.cloud.service.common.PostgresqlServiceInfo;
@@ -70,6 +73,10 @@ abstract public class StubCloudConnectorTest {
 	protected AmqpServiceInfo createRabbitService(String id) {
 		return new AmqpServiceInfo(id, "10.20.30.40", 1234, "username", "password", "vh");
 	}
+
+	protected CassandraServiceInfo createCassandraService(String id) {
+		return new CassandraServiceInfo(id, Arrays.asList("10.20.30.40"), 1234, "username", "password");
+	}
 	
 	protected RedisServiceInfo createRedisService(String id) {
 		return new RedisServiceInfo(id, "host", 1234, "password");

spring-cloud-spring-service-connector/src/test/java/org/springframework/cloud/config/java/CassandraClusterFactoryJavaConfigTest.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.config.java;
+
+import static org.hamcrest.CoreMatchers.*;
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+import org.springframework.cloud.service.ServiceInfo;
+import org.springframework.cloud.service.column.CassandraClusterConfig;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.SocketOptions;
+
+/**
+ * @author Mark Paluch
+ */
+public class CassandraClusterFactoryJavaConfigTest
+		extends AbstractServiceJavaConfigTest<Cluster> {
+
+	public CassandraClusterFactoryJavaConfigTest() {
+		super(CassandraClusterConfigWithId.class, CassandraClusterConfigWithoutId.class);
+	}
+
+	protected ServiceInfo createService(String id) {
+		return createCassandraService(id);
+	}
+
+	protected Class<Cluster> getConnectorType() {
+		return Cluster.class;
+	}
+
+	@Test
+	public void cloudRedisConnectionFactoryConfig() {
+		ApplicationContext testContext = getTestApplicationContext(
+				CassandraClusterConfigWithServiceConfig.class,
+				createService("my-service"));
+
+		Cluster connector = testContext.getBean("my-service",
+				getConnectorType());
+
+		assertThat(connector.getConfiguration().getSocketOptions().getSendBufferSize(),
+				is(12345));
+	}
+}
+
+class CassandraClusterConfigWithId extends AbstractCloudConfig {
+	@Bean(name = "my-service")
+	public Cluster testClusterFactory() {
+		return connectionFactory().cluster("my-service");
+	}
+}
+
+class CassandraClusterConfigWithoutId extends AbstractCloudConfig {
+	@Bean(name = "my-service")
+	public Cluster testClusterFactory() {
+		return connectionFactory().cluster();
+	}
+}
+
+class CassandraClusterConfigWithServiceConfig extends AbstractCloudConfig {
+	@Bean(name = "my-service")
+	public Cluster testClusterFactoryWithConfig() {
+
+		CassandraClusterConfig config = new CassandraClusterConfig();
+		SocketOptions socketOptions = new SocketOptions();
+		socketOptions.setSendBufferSize(12345);
+		config.setSocketOptions(socketOptions);
+
+		return connectionFactory().cluster("my-service", config);
+	}
+}

spring-cloud-spring-service-connector/src/test/java/org/springframework/cloud/config/xml/CassandraClusterXmlConfigTest.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.config.xml;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+import org.springframework.cloud.service.ServiceInfo;
+import org.springframework.context.ApplicationContext;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.policies.ConstantReconnectionPolicy;
+import com.datastax.driver.core.policies.DefaultRetryPolicy;
+import com.datastax.driver.core.policies.RoundRobinPolicy;
+
+/**
+ * @author Vinicius Carvalho
+ */
+public class CassandraClusterXmlConfigTest extends AbstractServiceXmlConfigTest<Cluster> {
+
+	@Test
+	public void cassandraSessionWithConfiguration() throws Exception {
+		ApplicationContext testContext = getTestApplicationContext(
+				"cloud-cassandra-with-config.xml", createService("my-service"));
+		Cluster cluster = testContext.getBean("cassandra-full-config",
+				getConnectorType());
+
+		assertNotNull(cluster.getConfiguration().getSocketOptions());
+		assertEquals(15000,
+				cluster.getConfiguration().getSocketOptions().getConnectTimeoutMillis());
+		assertTrue(DefaultRetryPolicy.class.isAssignableFrom(
+				cluster.getConfiguration().getPolicies().getRetryPolicy().getClass()));
+		assertTrue(RoundRobinPolicy.class.isAssignableFrom(cluster.getConfiguration()
+				.getPolicies().getLoadBalancingPolicy().getClass()));
+		assertTrue(ConstantReconnectionPolicy.class.isAssignableFrom(cluster
+				.getConfiguration().getPolicies().getReconnectionPolicy().getClass()));
+	}
+
+	@Override
+	protected String getWithServiceIdContextFileName() {
+		return "cloud-cassandra-with-service-id.xml";
+	}
+
+	@Override
+	protected String getWithoutServiceIdContextFileName() {
+		return "cloud-cassandra-without-service-id.xml";
+	}
+
+	@Override
+	protected ServiceInfo createService(String id) {
+		return createCassandraService(id);
+	}
+
+	@Override
+	protected Class<Cluster> getConnectorType() {
+		return Cluster.class;
+	}
+}

spring-cloud-spring-service-connector/src/test/java/org/springframework/cloud/config/xml/CloudAllServicesTest.java

@@ -4,6 +4,7 @@ import static org.junit.Assert.assertNotNull;
 
 import javax.sql.DataSource;
 
+import com.datastax.driver.core.Cluster;
 import org.junit.Test;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.beans.factory.BeanCreationException;
@@ -36,7 +37,8 @@ public class CloudAllServicesTest extends StubCloudConnectorTest {
 																   createPostgresqlService("postDb"),
 																   createMongoService("mongoDb"),
 																   createRedisService("redisDb"),
-																   createRabbitService("rabbit"));
+																   createRabbitService("rabbit"),
+																   createCassandraService("cassandra"));
 		
 		assertNotNull("Getting service by id", testContext.getBean("mysqlDb"));
 		assertNotNull("Getting service by id and type", testContext.getBean("mysqlDb", DataSource.class));		
@@ -51,7 +53,9 @@ public class CloudAllServicesTest extends StubCloudConnectorTest {
 		assertNotNull("Getting service by id and type", testContext.getBean("redisDb", RedisConnectionFactory.class));		
 
 		assertNotNull("Getting service by id", testContext.getBean("rabbit"));
-		assertNotNull("Getting service by id and type", testContext.getBean("rabbit", ConnectionFactory.class));		
+		assertNotNull("Getting service by id and type", testContext.getBean("rabbit", ConnectionFactory.class));
+
+		assertNotNull("Getting service by id", testContext.getBean("cassandra"));
+ 		assertNotNull("Getting service by id and type", testContext.getBean("cassandra", Cluster.class));
 	}
-	
 }

spring-cloud-spring-service-connector/src/test/java/org/springframework/cloud/service/column/CassandraClusterCreatorTest.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.service.column;
+
+import static org.hamcrest.CoreMatchers.*;
+import static org.junit.Assert.*;
+
+import java.util.Collections;
+
+import org.junit.Test;
+import org.springframework.cloud.service.common.CassandraServiceInfo;
+
+import com.datastax.driver.core.AuthProvider;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Configuration;
+import com.datastax.driver.core.PlainTextAuthProvider;
+import com.datastax.driver.core.PoolingOptions;
+import com.datastax.driver.core.ProtocolOptions;
+import com.datastax.driver.core.ProtocolVersion;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.policies.ConstantReconnectionPolicy;
+import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
+import com.datastax.driver.core.policies.Policies;
+import com.datastax.driver.core.policies.RoundRobinPolicy;
+
+/**
+ * @author Mark Paluch
+ */
+public class CassandraClusterCreatorTest {
+
+	CassandraClusterCreator creator = new CassandraClusterCreator();
+
+	@Test
+	public void shouldCreateCluster() throws Exception {
+
+		CassandraServiceInfo info = new CassandraServiceInfo("local",
+				Collections.singletonList("127.0.0.1"), 9142);
+
+		Cluster cluster = creator.create(info, null);
+
+		Configuration configuration = cluster.getConfiguration();
+
+		assertThat(configuration.getProtocolOptions().getAuthProvider(),
+				is(AuthProvider.NONE));
+	}
+
+	@Test
+	public void shouldCreateClusterWithAuthentication() throws Exception {
+
+		CassandraServiceInfo info = new CassandraServiceInfo("local",
+				Collections.singletonList("127.0.0.1"), 9142, "walter", "white");
+
+		Cluster cluster = creator.create(info, null);
+
+		Configuration configuration = cluster.getConfiguration();
+
+		assertThat(configuration.getProtocolOptions().getAuthProvider(),
+				is(instanceOf(PlainTextAuthProvider.class)));
+	}
+
+	@Test
+	public void shouldCreateClusterWithConfig() throws Exception {
+
+		CassandraServiceInfo info = new CassandraServiceInfo("local",
+				Collections.singletonList("127.0.0.1"), 9142);
+
+		CassandraClusterConfig config = new CassandraClusterConfig();
+		config.setCompression(ProtocolOptions.Compression.NONE);
+		config.setPoolingOptions(new PoolingOptions().setPoolTimeoutMillis(1234));
+		config.setQueryOptions(new QueryOptions());
+		config.setProtocolVersion(ProtocolVersion.NEWEST_SUPPORTED);
+		config.setLoadBalancingPolicy(new RoundRobinPolicy());
+		config.setReconnectionPolicy(new ConstantReconnectionPolicy(1));
+		config.setRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE);
+		config.setSocketOptions(new SocketOptions());
+
+		Cluster cluster = creator.create(info, config);
+
+		Configuration configuration = cluster.getConfiguration();
+
+		assertThat(configuration.getProtocolOptions().getCompression(),
+				is(config.getCompression()));
+		assertThat(configuration.getQueryOptions(), is(config.getQueryOptions()));
+		assertThat(configuration.getSocketOptions(), is(config.getSocketOptions()));
+
+		Policies policies = configuration.getPolicies();
+		assertThat(policies.getLoadBalancingPolicy(),
+				is(config.getLoadBalancingPolicy()));
+		assertThat(policies.getReconnectionPolicy(), is(config.getReconnectionPolicy()));
+		assertThat(policies.getRetryPolicy(), is(config.getRetryPolicy()));
+	}
+}

spring-cloud-spring-service-connector/src/test/java/org/springframework/cloud/service/column/CassandraClusterFactoryTest.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.service.column;
+
+import java.util.Arrays;
+
+import org.mockito.Mock;
+import org.springframework.cloud.service.AbstractCloudServiceConnectorFactoryTest;
+import org.springframework.cloud.service.ServiceConnectorConfig;
+import org.springframework.cloud.service.common.CassandraServiceInfo;
+
+import com.datastax.driver.core.Cluster;
+
+/**
+ * 
+ * @author Mark Paluch
+ */
+public class CassandraClusterFactoryTest extends
+		AbstractCloudServiceConnectorFactoryTest<CassandraClusterFactory, Cluster, CassandraServiceInfo> {
+	@Mock
+	Cluster mockConnector;
+
+	public CassandraClusterFactory createTestCloudServiceConnectorFactory(String id,
+			ServiceConnectorConfig config) {
+		return new CassandraClusterFactory(id, config);
+	}
+
+	public Class<Cluster> getConnectorType() {
+		return Cluster.class;
+	}
+
+	public Cluster getMockConnector() {
+		return mockConnector;
+	}
+
+	public CassandraServiceInfo getTestServiceInfo(String id) {
+		return new CassandraServiceInfo(id, Arrays.asList("127.0.0.1"), 9042, "user",
+				"pass");
+	}
+}

spring-cloud-spring-service-connector/src/test/resources/org/springframework/cloud/config/xml/cloud-all-services.xml

@@ -10,5 +10,6 @@
 	<cloud:mongo-db-factory service-name="mongoDb"/>
 	<cloud:redis-connection-factory service-name="redisDb"/>
 	<cloud:rabbit-connection-factory service-name="rabbit"/>
+	<cloud:cassandra-session-factory service-name="cassandra"/>
 	
 </beans>

spring-cloud-spring-service-connector/src/test/resources/org/springframework/cloud/config/xml/cloud-cassandra-with-config.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns:cloud="http://www.springframework.org/schema/cloud"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	http://www.springframework.org/schema/cloud http://www.springframework.org/schema/cloud/spring-cloud.xsd">
+
+    <bean id="loadBalancingPolicy" class="com.datastax.driver.core.policies.RoundRobinPolicy"/>
+    <bean id="reconnectionPolicy" class="com.datastax.driver.core.policies.ConstantReconnectionPolicy">
+        <constructor-arg index="0" value="2000"/>
+    </bean>
+    <bean id="retryPolicy" class="com.datastax.driver.core.policies.DefaultRetryPolicy"/>
+    <bean id="socketOptions" class="com.datastax.driver.core.SocketOptions">
+        <property name="keepAlive" value="true"/>
+        <property name="connectTimeoutMillis" value="15000"/>
+    </bean>
+
+    <cloud:cassandra-session-factory id="cassandra-full-config" service-name="my-service" >
+        <cloud:cassandra-options compression="NONE" retry-policy="retryPolicy"
+                                 loadbalancing-policy="loadBalancingPolicy" reconnection-policy="reconnectionPolicy"
+                                 socket-options="socketOptions"/>
+    </cloud:cassandra-session-factory>
+</beans>

spring-cloud-spring-service-connector/src/test/resources/org/springframework/cloud/config/xml/cloud-cassandra-with-service-id.xml

@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:cloud="http://www.springframework.org/schema/cloud"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	http://www.springframework.org/schema/cloud http://www.springframework.org/schema/cloud/spring-cloud.xsd">
+
+	<cloud:cassandra-session-factory service-name="my-service"/>
+
+</beans>

spring-cloud-spring-service-connector/src/test/resources/org/springframework/cloud/config/xml/cloud-cassandra-without-service-id.xml

@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:cloud="http://www.springframework.org/schema/cloud"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	http://www.springframework.org/schema/cloud http://www.springframework.org/schema/cloud/spring-cloud.xsd">
+
+	<cloud:cassandra-session-factory/>
+
+</beans>