Pluggable replicaplacement, endpointsnitch classes (take 2). patch by Sammy Yu; reviewed by jbellis for CASSANDRA-323

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@799065 13f79535-47bb-0310-9956-ffa450edef68

conf/storage-conf.xml

@@ -99,12 +99,23 @@
          clusters with a small number of nodes. -->
     <InitialToken></InitialToken>
 
-    <!-- RackAware: Setting this to true instructs Cassandra to try
+	
-         and place one replica in a different datacenter, and the
+    <!-- EndPointSnitch: Setting this to the class that implements IEndPointSnitch
-         others on different racks in the same one.  If you haven't
+	     which will see if two endpoints are in the same data center or on the same rack.
-         looked at the code for RackAwareStrategy, leave this off.
+         Out of the box, Cassandra provides
+         org.apache.cassandra.locator.EndPointSnitch
     -->
-    <RackAware>false</RackAware>
+    <EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
+
+    <!-- Strategy: Setting this to the class that implements IReplicaPlacementStrategy
+	     will change the way the node picker works.
+         Out of the box, Cassandra provides
+         org.apache.cassandra.locator.RackUnawareStrategy
+         org.apache.cassandra.locator.RackAwareStrategy 
+		 (place one replica in a different datacenter, and the
+         others on different racks in the same one.)
+    -->
+    <ReplicaPlacementStrategy>org.apache.cassandra.locator.RackUnawareStrategy</ReplicaPlacementStrategy>
 
     <!-- Number of replicas of the data-->
     <ReplicationFactor>1</ReplicationFactor>

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -31,6 +31,8 @@ import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.AsciiType;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.locator.IEndPointSnitch;
 import org.apache.cassandra.utils.FileUtils;
 import org.apache.cassandra.utils.XMLUtils;
 import org.w3c.dom.Node;
@@ -62,7 +64,6 @@ public class DatabaseDescriptor
     private static int currentIndex_ = 0;
     private static String logFileDirectory_;
     private static String bootstrapFileDirectory_;
-    private static boolean rackAware_ = false;
     private static int consistencyThreads_ = 4; // not configurable
     private static int concurrentReaders_ = 8;
     private static int concurrentWriters_ = 32;
@@ -87,7 +88,12 @@ public class DatabaseDescriptor
     */
     private static Map<String, Map<String, CFMetaData>> tableToCFMetaDataMap_;
     /* Hashing strategy Random or OPHF */
-    private static String partitionerClass_;
+    private static IPartitioner partitioner_;
+
+    private static IEndPointSnitch endPointSnitch_;
+
+    private static Class replicaPlacementStrategyClass_;
+
     /* if the size of columns or super-columns are more than this, indexing will kick in */
     private static int columnIndexSizeInKB_;
     /* Number of hours to keep a memtable in memory */
@@ -126,7 +132,7 @@ public class DatabaseDescriptor
         {
             configFileName_ = System.getProperty("storage-config") + File.separator + "storage-conf.xml";
             if (logger_.isDebugEnabled())
-              logger_.debug("Loading settings from " + configFileName_);
+                logger_.debug("Loading settings from " + configFileName_);
             XMLUtils xmlUtils = new XMLUtils(configFileName_);
 
             /* Cluster Name */
@@ -143,20 +149,37 @@ public class DatabaseDescriptor
             commitLogSyncDelay_ = Integer.valueOf(xmlUtils.getNodeValue("/Storage/CommitLogSyncDelay"));
 
             /* Hashing strategy */
-            partitionerClass_ = xmlUtils.getNodeValue("/Storage/Partitioner");
+            String partitionerClassName = xmlUtils.getNodeValue("/Storage/Partitioner");
-            try
+            if (partitionerClassName == null)
-            {
-                Class.forName(DatabaseDescriptor.getPartitionerClass());
-            }
-            catch (NullPointerException e)
             {
                 throw new ConfigurationException("Missing partitioner directive /Storage/Partitioner");
             }
+            try
+            {
+                Class cls = Class.forName(partitionerClassName);
+                partitioner_ = (IPartitioner) cls.getConstructor().newInstance();
+            }
             catch (ClassNotFoundException e)
             {
-                throw new ConfigurationException("Invalid partitioner class " + partitionerClass_);
+                throw new ConfigurationException("Invalid partitioner class " + partitionerClassName);
             }
 
+            /* end point snitch */
+            String endPointSnitchClassName = xmlUtils.getNodeValue("/Storage/EndPointSnitch");
+            if (endPointSnitchClassName == null)
+            {
+                throw new ConfigurationException("Missing endpointsnitch directive /Storage/EndPointSnitch");
+            }
+            try
+            {
+                Class cls = Class.forName(endPointSnitchClassName);
+                endPointSnitch_ = (IEndPointSnitch) cls.getConstructor().newInstance();
+            }
+            catch (ClassNotFoundException e)
+            {
+                throw new ConfigurationException("Invalid endpointsnitch class " + endPointSnitchClassName);
+            }
+            
             /* Callout location */
             calloutLocation_ = xmlUtils.getNodeValue("/Storage/CalloutLocation");
 
@@ -288,10 +311,20 @@ public class DatabaseDescriptor
             tableToCFMetaDataMap_ = new HashMap<String, Map<String, CFMetaData>>();
             tableKeysCachedFractions_ = new HashMap<String, Double>();
 
-            /* Rack Aware option */
+            /* See which replica placement strategy to use */
-            value = xmlUtils.getNodeValue("/Storage/RackAware");
+            String replicaPlacementStrategyClassName = xmlUtils.getNodeValue("/Storage/ReplicaPlacementStrategy");
-            if ( value != null )
+            if (replicaPlacementStrategyClassName == null)
-                rackAware_ = Boolean.parseBoolean(value);
+            {
+                throw new ConfigurationException("Missing replicaplacementstrategy directive /Storage/ReplicaPlacementStrategy");
+            }
+            try
+            {
+                replicaPlacementStrategyClass_ = Class.forName(replicaPlacementStrategyClassName);
+            }
+            catch (ClassNotFoundException e)
+            {
+                throw new ConfigurationException("Invalid replicaplacementstrategy class " + replicaPlacementStrategyClassName);
+            }
 
             /* Read the table related stuff from config */
             NodeList tables = xmlUtils.getRequestedNodeList("/Storage/Tables/Table");
@@ -542,9 +575,19 @@ public class DatabaseDescriptor
         return gcGraceInSeconds_;
     }
 
-    public static String getPartitionerClass()
+    public static IPartitioner getPartitioner()
     {
-        return partitionerClass_;
+        return partitioner_;
+    }
+    
+    public static IEndPointSnitch getEndPointSnitch()
+    {
+        return endPointSnitch_;
+    }
+
+    public static Class getReplicaPlacementStrategyClass()
+    {
+        return replicaPlacementStrategyClass_;
     }
     
     public static String getCalloutLocation()
@@ -760,11 +803,6 @@ public class DatabaseDescriptor
         logFileDirectory_ = logLocation;
     }
 
-    public static boolean isRackAware()
-    {
-        return rackAware_;
-    }
-
     public static Set<String> getSeeds()
     {
         return seeds_;

src/java/org/apache/cassandra/service/StorageService.java

@@ -111,6 +111,12 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
         FULL
     }
 
+    static
+    {
+        partitioner_ = DatabaseDescriptor.getPartitioner();
+    }
+
+
     public static class BootstrapInitiateDoneVerbHandler implements IVerbHandler
     {
         private static Logger logger_ = Logger.getLogger( BootstrapInitiateDoneVerbHandler.class );
@@ -203,8 +209,8 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
     {
         init();
         storageLoadBalancer_ = new StorageLoadBalancer(this);
-        endPointSnitch_ = new EndPointSnitch();
+        endPointSnitch_ = DatabaseDescriptor.getEndPointSnitch();
-        
+
         /* register the verb handlers */
         MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.tokenVerbHandler_, new TokenUpdateVerbHandler());
         MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.binaryVerbHandler_, new BinaryVerbHandler());
@@ -232,25 +238,18 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
         StageManager.registerStage(StorageService.readStage_,
                                    new MultiThreadedStage(StorageService.readStage_, DatabaseDescriptor.getConcurrentReaders()));
 
-        if ( DatabaseDescriptor.isRackAware() )
+        Class cls = DatabaseDescriptor.getReplicaPlacementStrategyClass();
-            nodePicker_ = new RackAwareStrategy(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort());
+        Class [] parameterTypes = new Class[] { TokenMetadata.class, IPartitioner.class, int.class, int.class};
-        else
-            nodePicker_ = new RackUnawareStrategy(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort());
-    }
-
-    static
-    {
         try
         {
-            Class cls = Class.forName(DatabaseDescriptor.getPartitionerClass());
+            nodePicker_ = (IReplicaPlacementStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort());
-            partitioner_ = (IPartitioner) cls.getConstructor().newInstance();
         }
         catch (Exception e)
         {
             throw new RuntimeException(e);
         }
     }
-    
+
     public void start() throws IOException
     {
         storageMetadata_ = SystemTable.initMetadata();

test/conf/storage-conf.xml

@@ -21,7 +21,8 @@
    <CommitLogSync>true</CommitLogSync>
    <CommitLogSyncDelay>1000</CommitLogSyncDelay>
    <Partitioner>org.apache.cassandra.dht.OrderPreservingPartitioner</Partitioner>
-   <RackAware>false</RackAware>
+   <EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
+   <ReplicaPlacementStrategy>org.apache.cassandra.locator.RackUnawareStrategy</ReplicaPlacementStrategy>
    <ReplicationFactor>1</ReplicationFactor>
    <RpcTimeoutInMillis>5000</RpcTimeoutInMillis>
    <ListenAddress>127.0.0.1</ListenAddress>