Make CassandraClientFactory a configurable instance.

Author: Tom May

config/elasticsearch.yml

@@ -1,5 +1,8 @@
 gateway:
   type: cassandra
+  cassandra:
+    host: localhost
+    port: 9160
 
 index:
   numberOfShards: 1

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

@@ -115,6 +115,8 @@ public class CassandraBlobStore extends AbstractComponent implements BlobStore {
 
     private final ESLogger logger = Loggers.getLogger(getClass());
 
+    private final CassandraClientFactory cassandraClientFactory;
+
     private final Executor executor;
 
     private final int bufferSizeInBytes; // XXX
@@ -123,11 +125,16 @@ public class CassandraBlobStore extends AbstractComponent implements BlobStore {
     public CassandraBlobStore(Settings settings, Executor executor) {
         super(settings);
 
+        String host = settings.get("host", "localhost");
+        int port = settings.getAsInt("port", 9160);
+        cassandraClientFactory = new CassandraClientFactory(host, port);
+
         this.executor = executor;
 
         this.bufferSizeInBytes = (int) settings.getAsBytesSize("buffer_size", new ByteSizeValue(100, ByteSizeUnit.KB)).bytes();
 
-        logger.debug("CassandraBlobStore executor: {} bufferSizeInBytes: {}", executor, bufferSizeInBytes);
+        logger.debug("CassandraBlobStore {}:{} executor: {} bufferSizeInBytes: {}",
+            host, port, executor, bufferSizeInBytes);
     }
 
     @Override public String toString() {
@@ -168,7 +175,7 @@ boolean blobExists(String blobPath, String blobName) {
         logger.debug("TODO blobExists {}", blobKey);
         try {
             Cassandra.Client client =
-                CassandraClientFactory.getCassandraClient();
+                cassandraClientFactory.getCassandraClient();
             try {
                 return client.get_count(
                     keySpace,
@@ -177,7 +184,7 @@ boolean blobExists(String blobPath, String blobName) {
                     ConsistencyLevel.QUORUM) != 0;
             }
             finally {
-                CassandraClientFactory.closeCassandraClient(client);
+                cassandraClientFactory.closeCassandraClient(client);
             }
         } catch (Exception e) {
             return false;
@@ -224,7 +231,7 @@ private boolean deleteBlobs(String blobPath, String... blobNames)
 
         Cassandra.Client client = null;
         try {
-            client = CassandraClientFactory.getCassandraClient();
+            client = cassandraClientFactory.getCassandraClient();
             client.batch_mutate(
                 keySpace, mutationMap, ConsistencyLevel.QUORUM);
             return true;
@@ -236,7 +243,7 @@ private boolean deleteBlobs(String blobPath, String... blobNames)
         }
         finally {
             if (client != null) {
-                CassandraClientFactory.closeCassandraClient(client);
+                cassandraClientFactory.closeCassandraClient(client);
             }
         }
     }
@@ -258,15 +265,15 @@ void readBlob(String blobPath, String blobName, final BlobContainer.ReadBlobList
             @Override public void run() {
                 Cassandra.Client client = null;
                 try {
-                    client = CassandraClientFactory.getCassandraClient();
+                    client = cassandraClientFactory.getCassandraClient();
                     readBlob(client, blobKey, listener);
                 }
                 catch (Exception ex) {
                     listener.onFailure(ex);
                 }
                 finally {
                     if (client != null) {
-                        CassandraClientFactory.closeCassandraClient(client);
+                        cassandraClientFactory.closeCassandraClient(client);
                     }
                 }
             }
@@ -291,7 +298,7 @@ private void readBlob(Cassandra.Client client, String blobKey, BlobContainer.Rea
     ImmutableMap<String, BlobMetaData> listBlobsByPrefix(String blobPath, @Nullable String blobNamePrefix) throws IOException {
         logger.debug("listBlobsByPrefix {}", blobKey(blobPath, blobNamePrefix));
         List<ColumnOrSuperColumn> columns;
-        Cassandra.Client client = CassandraClientFactory.getCassandraClient();
+        Cassandra.Client client = cassandraClientFactory.getCassandraClient();
         try {
             columns = client.get_slice(
                 keySpace,
@@ -317,7 +324,7 @@ ImmutableMap<String, BlobMetaData> listBlobsByPrefix(String blobPath, @Nullable
             throw new IOException("Cassandra get_slice on ???:??? failed", ex);
         }
         finally {
-            CassandraClientFactory.closeCassandraClient(client);
+            cassandraClientFactory.closeCassandraClient(client);
         }
 
         ImmutableMap.Builder<String, BlobMetaData> blobsBuilder = ImmutableMap.builder();
@@ -341,13 +348,13 @@ void writeBlob(final String blobPath, final String blobName, final InputStream i
             @Override public void run() {
                 try {
                     Cassandra.Client client =
-                        CassandraClientFactory.getCassandraClient();
+                        cassandraClientFactory.getCassandraClient();
                     try {
                         writeBlob(client, blobPath, blobName, is, sizeInBytes);
                         listener.onCompleted();
                     }
                     finally {
-                        CassandraClientFactory.closeCassandraClient(client);
+                        cassandraClientFactory.closeCassandraClient(client);
                     }
                 } catch (Exception e) {
                     listener.onFailure(e);

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

@@ -31,24 +31,33 @@
 import org.apache.cassandra.thrift.Cassandra;
 
 class CassandraClientFactory {
-    public static Cassandra.Client getCassandraClient()
+    private final String host;
+    private final int port;
+
+    public CassandraClientFactory(String host, int port) {
+        this.host = host;
+        this.port = port;
+    }
+
+    public Cassandra.Client getCassandraClient()
         throws IOException
     {
         TTransport transport =
-            new TFramedTransport(new TSocket("localhost", 9160));
+            new TFramedTransport(new TSocket(host, port));
         TProtocol protocol = new TBinaryProtocol(transport);
         Cassandra.Client client = new Cassandra.Client(protocol);
         try {
             transport.open();
         }
         catch (TTransportException ex) {
             throw new IOException(
-                "Cassandra transport.open to localhost:9160 failed", ex);
+                "Cassandra transport.open to " + host + ":" + port + " failed",
+                ex);
         }
         return client;
     }
 
-    public static void closeCassandraClient(Cassandra.Client client) {
+    public void closeCassandraClient(Cassandra.Client client) {
         client.getInputProtocol().getTransport().close();
     }
 }

plugins/cassandra/src/main/java/org/elasticsearch/gateway/cassandra/CassandraGateway.java

@@ -49,7 +49,7 @@ public class CassandraGateway extends BlobStoreGateway {
 
         logger.debug("using chunk_size [{}]", chunkSize);
 
-        initialize(new CassandraBlobStore(settings, threadPool.cached()), clusterName, chunkSize);
+        initialize(new CassandraBlobStore(componentSettings, threadPool.cached()), clusterName, chunkSize);
     }
 
     @Override public void close() throws ElasticSearchException {