diff --git a/conf/storage-conf.xml b/conf/storage-conf.xml index 9b7a472274..1115255542 100644 --- a/conf/storage-conf.xml +++ b/conf/storage-conf.xml @@ -99,12 +99,23 @@ clusters with a small number of nodes. --> - - false + org.apache.cassandra.locator.EndPointSnitch + + + org.apache.cassandra.locator.RackUnawareStrategy 1 diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index fe485b6edd..9899a82e67 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/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> 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"); - try - { - Class.forName(DatabaseDescriptor.getPartitionerClass()); - } - catch (NullPointerException e) + String partitionerClassName = xmlUtils.getNodeValue("/Storage/Partitioner"); + if (partitionerClassName == null) { 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>(); tableKeysCachedFractions_ = new HashMap(); - /* Rack Aware option */ - value = xmlUtils.getNodeValue("/Storage/RackAware"); - if ( value != null ) - rackAware_ = Boolean.parseBoolean(value); + /* See which replica placement strategy to use */ + String replicaPlacementStrategyClassName = xmlUtils.getNodeValue("/Storage/ReplicaPlacementStrategy"); + if (replicaPlacementStrategyClassName == null) + { + 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 getSeeds() { return seeds_; diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index d639bdea64..aba6f44e49 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/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() ) - nodePicker_ = new RackAwareStrategy(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort()); - else - nodePicker_ = new RackUnawareStrategy(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort()); - } - - static - { + Class cls = DatabaseDescriptor.getReplicaPlacementStrategyClass(); + Class [] parameterTypes = new Class[] { TokenMetadata.class, IPartitioner.class, int.class, int.class}; try { - Class cls = Class.forName(DatabaseDescriptor.getPartitionerClass()); - partitioner_ = (IPartitioner) cls.getConstructor().newInstance(); + nodePicker_ = (IReplicaPlacementStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort()); } catch (Exception e) { throw new RuntimeException(e); } } - + public void start() throws IOException { storageMetadata_ = SystemTable.initMetadata(); diff --git a/test/conf/storage-conf.xml b/test/conf/storage-conf.xml index 3fd2eea755..c72bb3fea5 100644 --- a/test/conf/storage-conf.xml +++ b/test/conf/storage-conf.xml @@ -21,7 +21,8 @@ true 1000 org.apache.cassandra.dht.OrderPreservingPartitioner - false + org.apache.cassandra.locator.EndPointSnitch + org.apache.cassandra.locator.RackUnawareStrategy 1 5000 127.0.0.1