Merge "support initial_token config option" from Asias

"This is needed by some tests to specify specific tokens."

database.cc

@@ -1471,6 +1471,18 @@ database::query_mutations(const query::read_command& cmd, const query::partition
     }
 }
 
+std::unordered_set<sstring> database::get_initial_tokens() {
+    std::unordered_set<sstring> tokens;
+    sstring tokens_string = get_config().initial_token();
+    try {
+        boost::split(tokens, tokens_string, boost::is_any_of(sstring(",")));
+    } catch (...) {
+        throw std::runtime_error(sprint("Unable to parse initial_token=%s", tokens_string));
+    }
+    tokens.erase("");
+    return tokens;
+}
+
 std::ostream& operator<<(std::ostream& out, const atomic_cell_or_collection& c) {
     return out << to_hex(c._data);
 }

database.hh

@@ -633,6 +633,8 @@ public:
     const logalloc::region_group& dirty_memory_region_group() const {
         return _dirty_memory_region_group;
     }
+
+    std::unordered_set<sstring> get_initial_tokens();
 };
 
 // FIXME: stub

db/config.hh

@@ -406,7 +406,7 @@ public:
             "The IP address a node tells other nodes in the cluster to contact it by. It allows public and private address to be different. For example, use the broadcast_address parameter in topologies where not all nodes have access to other nodes by their private IP addresses.\n" \
             "If your Cassandra cluster is deployed across multiple Amazon EC2 regions and you use the EC2MultiRegionSnitch , set the broadcast_address to public IP address of the node and the listen_address to the private IP."    \
     )   \
-    val(initial_token, sstring, /* N/A */, Unused,     \
+    val(initial_token, sstring, /* N/A */, Used,     \
             "Used in the single-node-per-token architecture, where a node owns exactly one contiguous range in the ring space. Setting this property overrides num_tokens.\n"   \
             "If you not using vnodes or have num_tokens set it to 1 or unspecified (#num_tokens), you should always specify this parameter when setting up a production cluster for the first time and when adding capacity. For more information, see this parameter in the Cassandra 1.1 Node and Cluster Configuration documentation.\n" \
             "This parameter can be used with num_tokens (vnodes ) in special cases such as Restoring from a snapshot." \

dht/boot_strapper.cc

@@ -71,33 +71,33 @@ future<> boot_strapper::bootstrap() {
 }
 
 std::unordered_set<token> boot_strapper::get_bootstrap_tokens(token_metadata metadata, database& db) {
-#if 0
-    Collection<String> initialTokens = DatabaseDescriptor.getInitialTokens();
+    auto initial_tokens = db.get_initial_tokens();
     // if user specified tokens, use those
-    if (initialTokens.size() > 0)
-    {
-        logger.debug("tokens manually specified as {}",  initialTokens);
-        List<Token> tokens = new ArrayList<Token>(initialTokens.size());
-        for (String tokenString : initialTokens)
-        {
-            Token token = StorageService.getPartitioner().getTokenFactory().fromString(tokenString);
-            if (metadata.getEndpoint(token) != null)
-                throw new ConfigurationException("Bootstrapping to existing token " + tokenString + " is not allowed (decommission/removenode the old node first).");
-            tokens.add(token);
+    if (initial_tokens.size() > 0) {
+        logger.debug("tokens manually specified as {}", initial_tokens);
+        std::unordered_set<token> tokens;
+        for (auto& token_string : initial_tokens) {
+            auto token = dht::global_partitioner().from_sstring(token_string);
+            if (metadata.get_endpoint(token)) {
+                throw std::runtime_error(sprint("Bootstrapping to existing token %s is not allowed (decommission/removenode the old node first).", token_string));
+            }
+            tokens.insert(token);
         }
+        logger.debug("Get manually specified bootstrap_tokens={}", tokens);
         return tokens;
     }
-#endif
+
     size_t num_tokens = db.get_config().num_tokens();
     if (num_tokens < 1) {
         throw std::runtime_error("num_tokens must be >= 1");
     }
 
-    // if (numTokens == 1)
-    //     logger.warn("Picking random token for a single vnode.  You should probably add more vnodes; failing that, you should probably specify the token manually");
+    if (num_tokens == 1) {
+        logger.warn("Picking random token for a single vnode.  You should probably add more vnodes; failing that, you should probably specify the token manually");
+    }
 
     auto tokens = get_random_tokens(metadata, num_tokens);
-    logger.debug("Get bootstrap_tokens={}", tokens);
+    logger.debug("Get random bootstrap_tokens={}", tokens);
     return tokens;
 }
 

dht/murmur3_partitioner.cc

@@ -75,6 +75,9 @@ token murmur3_partitioner::get_random_token() {
 }
 
 inline int64_t long_token(const token& t) {
+    if (t.is_minimum()) {
+        return std::numeric_limits<long>::min();
+    }
 
     if (t._data.size() != sizeof(int64_t)) {
         throw runtime_exception(sprint("Invalid token. Should have size %ld, has size %ld\n", sizeof(int64_t), t._data.size()));

service/storage_service.cc

@@ -115,11 +115,6 @@ std::experimental::optional<inet_address> get_replace_address() {
     return {};
 }
 
-std::unordered_set<sstring> get_initial_tokens() {
-    // FIXME: DatabaseDescriptor.getInitialTokens();
-    return std::unordered_set<sstring>();
-}
-
 bool get_property_join_ring() {
     // FIXME: Boolean.parseBoolean(System.getProperty("cassandra.join_ring", "true")))
     return true;
@@ -313,7 +308,7 @@ void storage_service::join_token_ring(int delay) {
         size_t num_tokens = _db.local().get_config().num_tokens();
         _bootstrap_tokens = db::system_keyspace::get_saved_tokens().get0();
         if (_bootstrap_tokens.empty()) {
-            auto initial_tokens = get_initial_tokens();
+            auto initial_tokens = _db.local().get_initial_tokens();
             if (initial_tokens.size() < 1) {
                 _bootstrap_tokens = boot_strapper::get_random_tokens(_token_metadata, num_tokens);
                 if (num_tokens == 1) {