Amended functions to make use of new AstyanaxClientFactory so that only one instance is created for each Cassandra cluster.

Author: James Tyack

src/main/java/com/hmsonline/storm/cassandra/trident/TridentCassandraLookupFunction.java

@@ -18,6 +18,7 @@
 import com.hmsonline.storm.cassandra.bolt.mapper.TridentColumnMapper;
 import com.hmsonline.storm.cassandra.bolt.mapper.TridentTupleMapper;
 import com.hmsonline.storm.cassandra.client.AstyanaxClient;
+import com.hmsonline.storm.cassandra.client.AstyanaxClientFactory;
 
 public class TridentCassandraLookupFunction<K, C, V> implements Function {
     private static final long serialVersionUID = 12132012L;
@@ -27,24 +28,35 @@ public class TridentCassandraLookupFunction<K, C, V> implements Function {
     private TridentColumnMapper<K, C, V> columnsMapper;
     private TridentTupleMapper<K, C, V> tupleMapper;
     private AstyanaxClient<K, C, V> client;
-    private String clientConfigKey;
+    private String cassandraClusterId;
 
     private Filter tupleFilter = null; // used to prevent processing for tuples
                                        // that should be skipped by the lookup
     private int numberOfOutputFields = 1; // used to emit when the incoming
                                           // tuple doesn't pass the filter check
     private boolean emitEmptyOnFailure = false;
 
-    public TridentCassandraLookupFunction(String clientConfigKey, TridentTupleMapper<K, C, V> tupleMapper,
+    /**
+     * @param cassandraClusterId Unique identifier for the Cassandra cluster
+     * @param tupleMapper
+     * @param columnMapper
+     */
+    public TridentCassandraLookupFunction(String cassandraClusterId, TridentTupleMapper<K, C, V> tupleMapper,
             TridentColumnMapper<K, C, V> columnMapper) {
         this.columnsMapper = columnMapper;
         this.tupleMapper = tupleMapper;
-        this.clientConfigKey = clientConfigKey;
+        this.cassandraClusterId = cassandraClusterId;
     }
 
-    public TridentCassandraLookupFunction(String clientConfigKey, TridentTupleMapper<K, C, V> tupleMapper,
+    /**
+     * @param cassandraClusterId Unique identifier for the Cassandra cluster
+     * @param tupleMapper
+     * @param columnMapper
+     * @param emitEmptyOnFailure
+     */
+    public TridentCassandraLookupFunction(String cassandraClusterId, TridentTupleMapper<K, C, V> tupleMapper,
             TridentColumnMapper<K, C, V> columnMapper, boolean emitEmptyOnFailure) {
-        this(clientConfigKey, tupleMapper, columnMapper);
+        this(cassandraClusterId, tupleMapper, columnMapper);
         this.emitEmptyOnFailure = emitEmptyOnFailure;
     }
 
@@ -63,10 +75,8 @@ public void setEmitEmptyOnFailure(boolean emitEmptyOnFailure) {
     @Override
     @SuppressWarnings({ "unchecked", "rawtypes" })
     public void prepare(Map stormConf, TridentOperationContext context) {
-        Map<String, Object> config = (Map<String, Object>) stormConf.get(this.clientConfigKey);
-
-        this.client = new AstyanaxClient<K, C, V>();
-        this.client.start(config);
+        Map<String, Object> config = (Map<String, Object>) stormConf.get(this.cassandraClusterId);
+        this.client = AstyanaxClientFactory.getInstance(cassandraClusterId, config);
     }
 
     @Override

src/main/java/com/hmsonline/storm/cassandra/trident/TridentCassandraWriteFunction.java

@@ -14,6 +14,7 @@
 
 import com.hmsonline.storm.cassandra.bolt.mapper.TridentTupleMapper;
 import com.hmsonline.storm.cassandra.client.AstyanaxClient;
+import com.hmsonline.storm.cassandra.client.AstyanaxClientFactory;
 import com.hmsonline.storm.cassandra.exceptions.StormCassandraException;
 import com.hmsonline.storm.cassandra.exceptions.TupleMappingException;
 
@@ -24,16 +25,16 @@ public class TridentCassandraWriteFunction<K, C, V> implements Function {
     protected TridentTupleMapper<K, C, V> tupleMapper;
     private AstyanaxClient<K, C, V> client;
 
-    private String clientConfigKey;
+    private String cassandraClusterId;
     private Object valueToEmit;
 
     public void setValueToEmitAfterWrite(Object valueToEmit) {
         this.valueToEmit = valueToEmit;
     }
 
-    public TridentCassandraWriteFunction(String clientConfigKey, TridentTupleMapper<K, C, V> tupleMapper) {
+    public TridentCassandraWriteFunction(String cassandraClusterId, TridentTupleMapper<K, C, V> tupleMapper) {
         this.tupleMapper = tupleMapper;
-        this.clientConfigKey = clientConfigKey;
+        this.cassandraClusterId = cassandraClusterId;
         this.valueToEmit = null;
     }
     public TridentCassandraWriteFunction(String clientConfigKey, TridentTupleMapper<K, C, V> tupleMapper,
@@ -45,9 +46,8 @@ public TridentCassandraWriteFunction(String clientConfigKey, TridentTupleMapper<
     @Override
     @SuppressWarnings({ "rawtypes", "unchecked" })
     public void prepare(Map stormConf, TridentOperationContext context) {
-        Map<String, Object> config = (Map<String, Object>) stormConf.get(this.clientConfigKey);
-        client = new AstyanaxClient<K, C, V>();
-        client.start(config);
+        Map<String, Object> config = (Map<String, Object>) stormConf.get(this.cassandraClusterId);
+        client = AstyanaxClientFactory.getInstance(cassandraClusterId, config);
     }
 
     @Override