Make Cassandra keyspace configurable.

Author: Tom May

config/elasticsearch.yml

@@ -3,6 +3,7 @@ gateway:
   cassandra:
     host: localhost
     port: 9160
+    keyspace: ElasticSearch
 
 index:
   numberOfShards: 1

plugins/cassandra/src/main/java/org/elasticsearch/cassandra/blobstore/CassandraBlobStore.java

@@ -109,12 +109,12 @@
  * @author Tom May ([email protected])
  */
 public class CassandraBlobStore extends AbstractComponent implements BlobStore {
-    private static final String keySpace = "ElasticSearch";
-
     private static final Charset utf8 = Charset.forName("UTF-8");
 
     private final ESLogger logger = Loggers.getLogger(getClass());
 
+    private final String keyspace;
+
     private final CassandraClientFactory cassandraClientFactory;
 
     private final Executor executor;
@@ -125,6 +125,8 @@ public class CassandraBlobStore extends AbstractComponent implements BlobStore {
     public CassandraBlobStore(Settings settings, Executor executor) {
         super(settings);
 
+        keyspace = settings.get("keyspace", "ElasticSearch");
+
         String host = settings.get("host", "localhost");
         int port = settings.getAsInt("port", 9160);
         cassandraClientFactory = new CassandraClientFactory(host, port);
@@ -178,7 +180,7 @@ boolean blobExists(String blobPath, String blobName) {
                 cassandraClientFactory.getCassandraClient();
             try {
                 return client.get_count(
-                    keySpace,
+                    keyspace,
                     blobKey,
                     new ColumnParent("Blobs"),
                     ConsistencyLevel.QUORUM) != 0;
@@ -233,7 +235,7 @@ private boolean deleteBlobs(String blobPath, String... blobNames)
         try {
             client = cassandraClientFactory.getCassandraClient();
             client.batch_mutate(
-                keySpace, mutationMap, ConsistencyLevel.QUORUM);
+                keyspace, mutationMap, ConsistencyLevel.QUORUM);
             return true;
         }
         catch (Exception e) {
@@ -284,7 +286,7 @@ private void readBlob(Cassandra.Client client, String blobKey, BlobContainer.Rea
         throws Exception
     {
         ColumnOrSuperColumn columnOrSuperColumn = client.get(
-            keySpace,
+            keyspace,
             blobKey,
             new ColumnPath("Blobs").setColumn(utf8.encode("data")),
             ConsistencyLevel.QUORUM);
@@ -301,7 +303,7 @@ ImmutableMap<String, BlobMetaData> listBlobsByPrefix(String blobPath, @Nullable
         Cassandra.Client client = cassandraClientFactory.getCassandraClient();
         try {
             columns = client.get_slice(
-                keySpace,
+                keyspace,
                 blobPath,
                 new ColumnParent("BlobNames"),
                 new SlicePredicate().setSlice_range(
@@ -409,7 +411,7 @@ private void writeBlob(Cassandra.Client client, String blobPath, String blobName
         mutationMap.put(blobPath, blobNamesMutationMap);
 
         client.batch_mutate(
-            keySpace, mutationMap, ConsistencyLevel.QUORUM);
+            keyspace, mutationMap, ConsistencyLevel.QUORUM);
     }
 
     private Mutation createInsert(String name, ByteBuffer value, long timestamp) {