diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGenerator.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGenerator.java
index eab6f7404..92644c214 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGenerator.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGenerator.java
@@ -32,6 +32,7 @@ import org.springframework.util.StringUtils;
  *
  * @author Matthew T. Adams
  * @author Alex Shvid
+ * @author Mark Paluch
  */
 public class CreateTableCqlGenerator extends TableOptionsCqlGenerator<TableSpecification<CreateTableSpecification>> {
 
@@ -43,6 +44,9 @@ public class CreateTableCqlGenerator extends TableOptionsCqlGenerator<TableSpeci
 		return new CreateTableCqlGenerator(specification).toCql();
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.cql.generator.TableOptionsCqlGenerator#spec()
+	 */
 	@Override
 	protected CreateTableSpecification spec() {
 		return (CreateTableSpecification) super.spec();
@@ -113,7 +117,7 @@ public class CreateTableCqlGenerator extends TableOptionsCqlGenerator<TableSpeci
 		// begin option clause
 		Map<String, Object> options = spec().getOptions();
 
-		if (!options.isEmpty()) {
+		if (!options.isEmpty() || StringUtils.hasText(ordering)) {
 
 			// option preamble
 			boolean first = true;
@@ -164,8 +168,8 @@ public class CreateTableCqlGenerator extends TableOptionsCqlGenerator<TableSpeci
 		for (ColumnSpecification col : columns) {
 
 			if (col.getOrdering() != null) { // then ordering specified
-				if (StringUtils.isEmpty(ordering)) { // then initialize ordering clause
-					ordering = new StringBuilder().append("CLUSTERING ORDER BY (");
+				if (!StringUtils.hasText(ordering)) { // then initialize ordering clause
+					ordering.append("CLUSTERING ORDER BY (");
 				}
 				if (first) {
 					first = false;
@@ -193,7 +197,6 @@ public class CreateTableCqlGenerator extends TableOptionsCqlGenerator<TableSpeci
 				str.append(", ");
 			}
 			str.append(col.getName());
-
 		}
 	}
 }
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGeneratorIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGeneratorIntegrationTests.java
index 778fa2d1d..85b1c06e7 100755
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGeneratorIntegrationTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGeneratorIntegrationTests.java
@@ -15,57 +15,74 @@
  */
 package org.springframework.data.cassandra.core.cql.generator;
 
+import static org.assertj.core.api.Assertions.*;
+
+import org.junit.Before;
 import org.junit.Test;
-import org.springframework.data.cassandra.core.cql.generator.CreateTableCqlGeneratorUnitTests.BasicTest;
-import org.springframework.data.cassandra.core.cql.generator.CreateTableCqlGeneratorUnitTests.CompositePartitionKeyTest;
-import org.springframework.data.cassandra.core.cql.generator.CreateTableCqlGeneratorUnitTests.CreateTableTest;
+import org.springframework.data.cassandra.core.cql.Ordering;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification;
+import org.springframework.data.cassandra.core.cql.keyspace.TableOption;
 import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
 
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.TableMetadata;
+
 /**
- * Integration tests that reuse unit tests.
+ * Integration tests for {@link CreateTableCqlGenerator}.
  *
  * @author Matthew T. Adams
  * @author Oliver Gierke
+ * @author Mark Paluch
  */
-public class CreateTableCqlGeneratorIntegrationTests {
+public class CreateTableCqlGeneratorIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
 
-	/**
-	 * Integration test base class that knows how to do everything except instantiate the concrete unit test type T.
-	 *
-	 * @author Matthew T. Adams
-	 * @param <T> The concrete unit test class to which this integration test corresponds.
-	 */
-	public static abstract class Base<T extends CreateTableTest> extends AbstractKeyspaceCreatingIntegrationTest {
+	@Before
+	public void setUp() {
 
-		T unit;
-
-		public abstract T unit();
-
-		@Test
-		public void test() {
-
-			unit = unit();
-			unit.prepare();
-
-			session.execute(unit.cql);
-
-			CqlTableSpecificationAssertions.assertTable(unit.specification, keyspace, session);
-		}
+		session.execute("DROP TABLE IF EXISTS person;");
+		session.execute("DROP TABLE IF EXISTS address;");
 	}
 
-	public static class BasicIntegrationTest extends Base<BasicTest> {
+	@Test // DATACASS-518
+	public void shouldGenerateSimpleTable() {
 
-		@Override
-		public BasicTest unit() {
-			return new BasicTest();
-		}
+		CreateTableSpecification table = CreateTableSpecification.createTable("person") //
+				.partitionKeyColumn("id", DataType.ascii()) //
+				.clusteredKeyColumn("date_of_birth", DataType.date()) //
+				.column("name", DataType.ascii());
+
+		session.execute(CreateTableCqlGenerator.toCql(table));
 	}
 
-	public static class CompositePartitionKeyIntegrationTest extends Base<CompositePartitionKeyTest> {
+	@Test // DATACASS-518
+	public void shouldGenerateTableWithClusterKeyOrdering() {
 
-		@Override
-		public CompositePartitionKeyTest unit() {
-			return new CompositePartitionKeyTest();
-		}
+		CreateTableSpecification table = CreateTableSpecification.createTable("person") //
+				.partitionKeyColumn("id", DataType.ascii()) //
+				.partitionKeyColumn("country", DataType.ascii()) //
+				.clusteredKeyColumn("date_of_birth", DataType.date(), Ordering.ASCENDING) //
+				.clusteredKeyColumn("age", DataType.smallint()) //
+				.column("name", DataType.ascii());
+
+		TableMetadata person = cluster.getMetadata().getKeyspace(getKeyspace()).getTable("person");
+		assertThat(person.getPartitionKey()).hasSize(2);
+		assertThat(person.getClusteringColumns()).hasSize(2);
+
+		session.execute(CreateTableCqlGenerator.toCql(table));
+	}
+
+	@Test // DATACASS-518
+	public void shouldGenerateTableWithClusterKeyAndOptions() {
+
+		CreateTableSpecification table = CreateTableSpecification.createTable("person") //
+				.partitionKeyColumn("id", DataType.ascii()) //
+				.clusteredKeyColumn("date_of_birth", DataType.date(), Ordering.ASCENDING) //
+				.column("name", DataType.ascii()).with(TableOption.COMPACT_STORAGE);
+
+		session.execute(CreateTableCqlGenerator.toCql(table));
+
+		TableMetadata person = cluster.getMetadata().getKeyspace(getKeyspace()).getTable("person");
+		assertThat(person.getPartitionKey()).hasSize(1);
+		assertThat(person.getClusteringColumns()).hasSize(1);
 	}
 }
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGeneratorUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGeneratorUnitTests.java
index 6d723dfa5..ec75a672b 100755
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGeneratorUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGeneratorUnitTests.java
@@ -30,6 +30,7 @@ import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.cassandra.core.cql.Ordering;
 import org.springframework.data.cassandra.core.cql.ReservedKeyword;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.Option;
@@ -46,11 +47,40 @@ import com.datastax.driver.core.DataType;
  *
  * @author Matthew T. Adams
  * @author David Webb
+ * @author Mark Paluch
  */
 public class CreateTableCqlGeneratorUnitTests {
 
 	private static final Logger log = LoggerFactory.getLogger(CreateTableCqlGeneratorUnitTests.class);
 
+	@Test // DATACASS-518
+	public void createTableWithOrderedClustering() {
+
+		CreateTableSpecification table = CreateTableSpecification.createTable("person") //
+				.partitionKeyColumn("id", DataType.ascii()) //
+				.clusteredKeyColumn("date_of_birth", DataType.date(), Ordering.ASCENDING) //
+				.column("name", DataType.ascii());
+
+		assertThat(CreateTableCqlGenerator.toCql(table))
+				.isEqualTo("CREATE TABLE person (id ascii, date_of_birth date, name ascii, " //
+						+ "PRIMARY KEY (id, date_of_birth)) " //
+						+ "WITH CLUSTERING ORDER BY (date_of_birth ASC);");
+	}
+
+	@Test // DATACASS-518
+	public void createTableWithOrderedClusteringAndOptions() {
+
+		CreateTableSpecification table = CreateTableSpecification.createTable("person") //
+				.partitionKeyColumn("id", DataType.ascii()) //
+				.clusteredKeyColumn("date_of_birth", DataType.date(), Ordering.ASCENDING) //
+				.column("name", DataType.ascii()).with(TableOption.COMPACT_STORAGE);
+
+		assertThat(CreateTableCqlGenerator.toCql(table))
+				.isEqualTo("CREATE TABLE person (id ascii, date_of_birth date, name ascii, " //
+						+ "PRIMARY KEY (id, date_of_birth)) " //
+						+ "WITH CLUSTERING ORDER BY (date_of_birth ASC) AND COMPACT STORAGE;");
+	}
+
 	/**
 	 * Asserts that the preamble is first & correctly formatted in the given CQL string.
 	 */