ColumnFamilyStore: Cleanup and adding missing implementation

This patch adds the implementation for the ColumnFamilyStore by adding
calls to the relevent API that define in column_family.json

This patch also do some clean up, by removing leading underscore from
parameter names and trailing white spaces.

Signed-off-by: Amnon Heiman <amnon@cloudius-systems.com>

src/main/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -16,7 +16,6 @@
  * limitations under the License.
  */
 
-
 /*
  * Copyright 2015 Cloudius Systems
  *
@@ -31,8 +30,10 @@ import java.util.concurrent.*;
 import javax.json.JsonArray;
 import javax.json.JsonObject;
 import javax.management.*;
+import javax.ws.rs.core.MultivaluedMap;
 
 import com.cloudius.urchin.api.APIClient;
+import com.sun.jersey.core.util.MultivaluedMapImpl;
 
 public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
     private static final java.util.logging.Logger logger = java.util.logging.Logger
@@ -56,10 +57,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
         timer.scheduleAtFixedRate(taskToExecute, 100, INTERVAL);
     }
 
-    public ColumnFamilyStore(String _type, String _keyspace, String _name) {
+    public ColumnFamilyStore(String type, String keyspace, String name) {
-        type = _type;
+        this.type = type;
-        keyspace = _keyspace;
+        this.keyspace = keyspace;
-        name = _name;
+        this.name = name;
         mbeanName = getName(type, keyspace, name);
         try {
             MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
@@ -70,6 +71,15 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
         }
     }
 
+    /**
+     * Get the column family name in the API format
+     *
+     * @return
+     */
+    public String getCFName() {
+        return keyspace + ":" + name;
+    }
+
     private static String getName(String type, String keyspace, String name) {
         return "org.apache.cassandra.db:type=" + type + ",keyspace=" + keyspace
                 + ",columnfamily=" + name;
@@ -322,6 +332,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
     public void forceMajorCompaction() throws ExecutionException,
             InterruptedException {
         log(" forceMajorCompaction() throws ExecutionException, InterruptedException");
+        c.post("column_family/major_compaction/" + getCFName());
     }
 
     /**
@@ -431,7 +442,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public int getMinimumCompactionThreshold() {
         log(" getMinimumCompactionThreshold()");
-        return c.getIntValue("");
+        return c.getIntValue("column_family/minimum_compaction/" + getCFName());
     }
 
     /**
@@ -439,6 +450,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public void setMinimumCompactionThreshold(int threshold) {
         log(" setMinimumCompactionThreshold(int threshold)");
+        MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
+        queryParams.add("value", Integer.toString(threshold));
+        c.post("column_family/minimum_compaction/" + getCFName(), queryParams);
     }
 
     /**
@@ -446,7 +460,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public int getMaximumCompactionThreshold() {
         log(" getMaximumCompactionThreshold()");
-        return c.getIntValue("");
+        return c.getIntValue("column_family/maximum_compaction/" + getCFName());
     }
 
     /**
@@ -455,6 +469,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public void setCompactionThresholds(int minThreshold, int maxThreshold) {
         log(" setCompactionThresholds(int minThreshold, int maxThreshold)");
+        MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
+        queryParams.add("minimum", Integer.toString(minThreshold));
+        queryParams.add("maximum", Integer.toString(maxThreshold));
+        c.post("column_family/compaction" + getCFName(), queryParams);
     }
 
     /**
@@ -462,6 +480,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public void setMaximumCompactionThreshold(int threshold) {
         log(" setMaximumCompactionThreshold(int threshold)");
+        MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
+        queryParams.add("value", Integer.toString(threshold));
+        c.post("column_family/maximum_compaction/" + getCFName(), queryParams);
     }
 
     /**
@@ -472,6 +493,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public void setCompactionStrategyClass(String className) {
         log(" setCompactionStrategyClass(String className)");
+        MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
+        queryParams.add("class_name", className);
+        c.post("column_family/compaction_strategy/" + getCFName(), queryParams);
     }
 
     /**
@@ -479,7 +503,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public String getCompactionStrategyClass() {
         log(" getCompactionStrategyClass()");
-        return c.getStringValue("");
+        return c.getStringValue("column_family/compaction_strategy/"
+                + getCFName());
     }
 
     /**
@@ -487,7 +512,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public Map<String, String> getCompressionParameters() {
         log(" getCompressionParameters()");
-        return c.getMapStrValue("");
+        return c.getMapStrValue("column_family/compression_parameters/"
+                + getCFName());
     }
 
     /**
@@ -498,6 +524,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public void setCompressionParameters(Map<String, String> opts) {
         log(" setCompressionParameters(Map<String,String> opts)");
+        MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
+        queryParams.add("opts", APIClient.mapToString(opts));
+        c.post("column_family/compression_parameters/" + getCFName(),
+                queryParams);
     }
 
     /**
@@ -505,11 +535,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public void setCrcCheckChance(double crcCheckChance) {
         log(" setCrcCheckChance(double crcCheckChance)");
+        MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
+        queryParams.add("check_chance", Double.toString(crcCheckChance));
+        c.post("column_family/crc_check_chance/" + getCFName(), queryParams);
     }
 
     public boolean isAutoCompactionDisabled() {
         log(" isAutoCompactionDisabled()");
-        return c.getBooleanValue("");
+        return c.getBooleanValue("column_family/autocompaction/" + getCFName());
     }
 
     /** Number of tombstoned cells retreived during the last slicequery */
@@ -528,7 +561,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
 
     public long estimateKeys() {
         log(" estimateKeys()");
-        return c.getLongValue("");
+        return c.getLongValue("column_family/estimate_keys/" + getCFName());
     }
 
     /**
@@ -565,7 +598,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public List<String> getBuiltIndexes() {
         log(" getBuiltIndexes()");
-        return c.getListStrValue("");
+        return c.getListStrValue("column_family/built_indexes/" + getCFName());
     }
 
     /**
@@ -576,7 +609,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public List<String> getSSTablesForKey(String key) {
         log(" getSSTablesForKey(String key)");
-        return c.getListStrValue("");
+        MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
+        queryParams.add("key", key);
+        return c.getListStrValue(
+                "column_family/sstables/by_key/" + getCFName(), queryParams);
     }
 
     /**
@@ -585,6 +621,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public void loadNewSSTables() {
         log(" loadNewSSTables()");
+        c.post("column_family/sstable/" + getCFName());
     }
 
     /**
@@ -593,7 +630,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public int getUnleveledSSTables() {
         log(" getUnleveledSSTables()");
-        return c.getIntValue("");
+        return c.getIntValue("column_family/sstables/unleveled/" + getCFName());
     }
 
     /**
@@ -603,7 +640,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public int[] getSSTableCountPerLevel() {
         log(" getSSTableCountPerLevel()");
-        return c.getIntArrValue("");
+        return c.getIntArrValue("column_family/sstables/per_level/"
+                + getCFName());
     }
 
     /**
@@ -614,7 +652,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public double getDroppableTombstoneRatio() {
         log(" getDroppableTombstoneRatio()");
-        return c.getDoubleValue("");
+        return c.getDoubleValue("column_family/droppable_ratio/" + getCFName());
     }
 
     /**
@@ -623,7 +661,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
      */
     public long trueSnapshotsSize() {
         log(" trueSnapshotsSize()");
-        return c.getLongValue("");
+        return c.getLongValue("column_family/snapshots_size/" + getCFName());
     }
 
 }