From 73c59865f7d65b318f01fa81b0cf573238fe3098 Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@scylladb.com>
Date: Tue, 25 May 2021 17:36:23 +0300
Subject: [PATCH 1/8] raft: implement is_transient_error()

Add a helper to classify Raft exceptions as transient.
---
 raft/raft.hh | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/raft/raft.hh b/raft/raft.hh
index 37c1658f6b..99a9f12c4d 100644
--- a/raft/raft.hh
+++ b/raft/raft.hh
@@ -223,6 +223,16 @@ struct config_error : public error {
     using error::error;
 };
 
+// True if a failure to execute a Raft operation can be re-tried,
+// perhaps with a different server.
+inline bool is_transient_error(const std::exception& e) {
+    return dynamic_cast<const not_a_leader*>(&e) ||
+           dynamic_cast<const commit_status_unknown*>(&e) ||
+           dynamic_cast<const dropped_entry*>(&e) ||
+           dynamic_cast<const conf_change_in_progress*>(&e) ||
+           dynamic_cast<const stopped_error*>(&e);
+}
+
 struct snapshot {
     // Index and term of last entry in the snapshot
     index_t idx = index_t(0);

From b8fa6c6e9cd2916e0840b56d284f42939cd9a276 Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@scylladb.com>
Date: Thu, 13 May 2021 18:50:49 +0300
Subject: [PATCH 2/8] raft: (service) properly handle failure to add a server

future.get() is not available outside thread context
and co_await is not available inside catch (...) block.
---
 service/raft/raft_services.cc | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/service/raft/raft_services.cc b/service/raft/raft_services.cc
index 007b448c1b..1ab93d9dc0 100644
--- a/service/raft/raft_services.cc
+++ b/service/raft/raft_services.cc
@@ -133,15 +133,19 @@ seastar::future<> raft_services::init() {
     // and starts an election, so RPC must be ready by
     // then to send VoteRequest messages.
     init_rpc_verbs();
-    auto uninit_rpc_verbs = defer([this] { this->uninit_rpc_verbs().get(); });
     // schema raft server instance always resides on shard 0
-    if (this_shard_id() == 0) {
-        // FIXME: Server id will change each time scylla server restarts,
-        // need to persist it or find a deterministic way to compute!
+    if (this_shard_id() != 0) {
+        co_return;
+    }
+    co_await [this] () -> future<> {
         raft::server_id id = raft::server_id::create_random_id();
         co_await add_server(id, create_schema_server(id));
-    }
-    uninit_rpc_verbs.cancel();
+    }().handle_exception([this] (auto ep) -> future<> {
+        // co_await is not allowed inside catch (...), use
+        // handle_exception instead.
+        co_await uninit_rpc_verbs();
+        std::rethrow_exception(ep);
+    });
 }
 
 seastar::future<> raft_services::uninit() {

From 845ff9f3446095945a912dcb118e4fa0a534cc78 Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@scylladb.com>
Date: Thu, 13 May 2021 18:52:18 +0300
Subject: [PATCH 3/8] raft: (service) permit some scylla.raft cells to be empty

When loading raft state from scylla.raft, permit some cells
to be empty. Indeed, the server is not obliged to persist
all vote, term, snapshot once it starts. And the log can be
empty.
---
 service/raft/raft_sys_table_storage.cc | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/service/raft/raft_sys_table_storage.cc b/service/raft/raft_sys_table_storage.cc
index 59469fdb7d..18e3de63f8 100644
--- a/service/raft/raft_sys_table_storage.cc
+++ b/service/raft/raft_sys_table_storage.cc
@@ -66,8 +66,8 @@ future<std::pair<raft::term_t, raft::server_id>> raft_sys_table_storage::load_te
         co_return std::pair(raft::term_t(), raft::server_id());
     }
     const auto& static_row = rs->one();
-    raft::term_t vote_term = raft::term_t(static_row.get_as<int64_t>("vote_term"));
-    raft::server_id vote{static_row.get_as<utils::UUID>("vote")};
+    raft::term_t vote_term = raft::term_t(static_row.get_or<int64_t>("vote_term", raft::term_t{}));
+    raft::server_id vote{static_row.get_or<utils::UUID>("vote", raft::server_id{}.id)};
     co_return std::pair(vote_term, vote);
 }
 
@@ -77,6 +77,11 @@ future<raft::log_entries> raft_sys_table_storage::load_log() {
 
     raft::log_entries log;
     for (const cql3::untyped_result_set_row& row : *rs) {
+        if (!row.has("data")) {
+            // The partition only contains static cells, the log
+            // is empty.
+            break;
+        }
         raft::term_t term = raft::term_t(row.get_as<int64_t>("term"));
         raft::index_t idx = raft::index_t(row.get_as<int64_t>("index"));
         auto raw_data = row.get_blob("data");
@@ -95,7 +100,7 @@ future<raft::log_entries> raft_sys_table_storage::load_log() {
 future<raft::snapshot> raft_sys_table_storage::load_snapshot() {
     static const auto load_id_cql = format("SELECT snapshot_id FROM system.{} WHERE group_id = ?", db::system_keyspace::RAFT);
     ::shared_ptr<cql3::untyped_result_set> id_rs = co_await _qp.execute_internal(load_id_cql, {_group_id.id});
-    if (id_rs->empty()) {
+    if (id_rs->empty() || !id_rs->one().has("snapshot_id")) {
         co_return raft::snapshot();
     }
     const auto& id_row = id_rs->one(); // should be only one row since snapshot_id column is static

From 867440f08053c4700e4313670bb4f7bdefac8b0a Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@scylladb.com>
Date: Thu, 13 May 2021 19:14:32 +0300
Subject: [PATCH 4/8] raft: (service) don't throw from schema state machine

It's now started as Scylla starts, and state machine failure
leads to panic at start.
---
 service/raft/schema_raft_state_machine.cc | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/service/raft/schema_raft_state_machine.cc b/service/raft/schema_raft_state_machine.cc
index 6f91181f51..09cd842922 100644
--- a/service/raft/schema_raft_state_machine.cc
+++ b/service/raft/schema_raft_state_machine.cc
@@ -21,19 +21,19 @@
 #include "service/raft/schema_raft_state_machine.hh"
 
 future<> schema_raft_state_machine::apply(std::vector<raft::command_cref> command) {
-    throw std::runtime_error("Not implemented");
+    return make_ready_future<>();
 }
 
 future<raft::snapshot_id> schema_raft_state_machine::take_snapshot() {
-    throw std::runtime_error("Not implemented");
+    return make_ready_future<raft::snapshot_id>(raft::snapshot_id::create_random_id());
 }
 
 void schema_raft_state_machine::drop_snapshot(raft::snapshot_id id) {
-    throw std::runtime_error("Not implemented");
+    (void) id;
 }
 
 future<> schema_raft_state_machine::load_snapshot(raft::snapshot_id id) {
-    throw std::runtime_error("Not implemented");
+    return make_ready_future<>();
 }
 
 future<> schema_raft_state_machine::abort() {

From 631c89e1a6631c81f7797ff96481855ac5c3e6ee Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@scylladb.com>
Date: Fri, 21 May 2021 16:02:42 +0300
Subject: [PATCH 5/8] raft: (server) implement
 raft::server::get_configuration()

raft::server::set_configuration() is useless on
application level if we can't query the previous configuration.
---
 raft/server.cc | 6 ++++++
 raft/server.hh | 3 +++
 2 files changed, 9 insertions(+)

diff --git a/raft/server.cc b/raft/server.cc
index 345127a277..20c96d9c4d 100644
--- a/raft/server.cc
+++ b/raft/server.cc
@@ -65,6 +65,7 @@ public:
     future<> add_entry(command command, wait_type type);
     future<snapshot_reply> apply_snapshot(server_id from, install_snapshot snp) override;
     future<> set_configuration(server_address_set c_new) override;
+    raft::configuration get_configuration() const override;
     future<> start() override;
     future<> abort() override;
     term_t get_current_term() const override;
@@ -719,6 +720,11 @@ future<> server_impl::set_configuration(server_address_set c_new) {
     co_return co_await add_entry_internal(raft::configuration{std::move(c_new)}, wait_type::committed);
 }
 
+raft::configuration
+server_impl::get_configuration() const {
+    return _fsm->get_configuration();
+}
+
 void server_impl::register_metrics() {
     namespace sm = seastar::metrics;
     _metrics.add_group("raft", {
diff --git a/raft/server.hh b/raft/server.hh
index 2313083244..20a618a8a2 100644
--- a/raft/server.hh
+++ b/raft/server.hh
@@ -73,6 +73,9 @@ public:
     // otherwise throws conf_change_in_progress exception.
     virtual future<> set_configuration(server_address_set c_new) = 0;
 
+    // Return the currently known configuration
+    virtual raft::configuration get_configuration() const = 0;
+
     // Load persisted state and start background work that needs
     // to run for this Raft server to function; The object cannot
     // be used until the returned future is resolved.

From c67c77ed03922cbd973190fcfee6d60ec155b890 Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@scylladb.com>
Date: Sat, 22 May 2021 07:31:57 +0300
Subject: [PATCH 6/8] raft: (server) wait for configuration transition to
 complete

By default, wait for the server to leave the joint configuration
when making a configuration change.

When assembling a fresh cluster Scylla may run a series of
configuration changes. These changes would all go through the same
leader and serialize in the critical section around server::cas().

Unless this critical section protects the complete transition from
C_old configuration to C_new, after the first configuration
is committed, the second may fail with exception that a configuration
change is in progress. The topology changes layer should handle
this exception, however, this may introduce either unpleasant
delays into cluster assembly (i.e. if we sleep before retry), or
a busy-wait/thundering herd situation, when all nodes are
retrying their configuration changes.

So let's be nice and wait for a full transition in
server::set_configuration().
---
 raft/server.cc | 10 +++++++++-
 raft/server.hh |  8 ++++++++
 2 files changed, 17 insertions(+), 1 deletion(-)

diff --git a/raft/server.cc b/raft/server.cc
index 20c96d9c4d..8bb009c71e 100644
--- a/raft/server.cc
+++ b/raft/server.cc
@@ -717,7 +717,15 @@ future<> server_impl::set_configuration(server_address_set c_new) {
         co_return;
     }
     _stats.add_config++;
-    co_return co_await add_entry_internal(raft::configuration{std::move(c_new)}, wait_type::committed);
+    co_await add_entry_internal(raft::configuration{std::move(c_new)}, wait_type::committed);
+     // Above we co_wait that the joint configuration is committed.
+     // Immediately, without yield, once the FSM discovers
+     // this, it appends non-joint entry. Hence,
+     // at this point in execution, non-joint entry is in the log.
+     // By waiting for a follow up dummy to get committed, we
+     // automatically wait for the non-joint entry to get
+     // committed.
+    co_await add_entry_internal(log_entry::dummy(), wait_type::committed);
 }
 
 raft::configuration
diff --git a/raft/server.hh b/raft/server.hh
index 20a618a8a2..c2fdb14555 100644
--- a/raft/server.hh
+++ b/raft/server.hh
@@ -71,6 +71,14 @@ public:
     // Can be called on a leader only, otherwise throws not_a_leader.
     // Cannot be called until previous set_configuration() completes
     // otherwise throws conf_change_in_progress exception.
+    //
+    // Waits until configuration completes, i.e. the server left the joint
+    // configuration. The server will apply a dummy entry to
+    // make sure this happens.
+    //
+    // Note: committing a dummy entry extends the opportunity for
+    // uncertainty, thus commit_status_unknown exception may be
+    // returned even in case of a successful config change.
     virtual future<> set_configuration(server_address_set c_new) = 0;
 
     // Return the currently known configuration

From 9c93d77e74018ab90f8e5ee34e8feaa3fd9e6b44 Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@scylladb.com>
Date: Tue, 25 May 2021 21:26:53 +0300
Subject: [PATCH 7/8] system_keyspace: raft.group_id and
 raft_snapshots.group_id are TIMEUUID

Fix a bug in definitions of system.raft, system.raft_snapshots,
group_id is TIMEUUID, not long.
---
 db/system_keyspace.cc                     | 4 ++--
 service/raft/raft_sys_table_storage.cc    | 2 +-
 service/raft/schema_raft_state_machine.cc | 2 ++
 service/raft/schema_raft_state_machine.hh | 3 ++-
 test/boost/raft_sys_table_storage_test.cc | 4 ++--
 5 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/db/system_keyspace.cc b/db/system_keyspace.cc
index 5993d5662f..04364ecba0 100644
--- a/db/system_keyspace.cc
+++ b/db/system_keyspace.cc
@@ -220,7 +220,7 @@ schema_ptr raft() {
     static thread_local auto schema = [] {
         auto id = generate_legacy_id(NAME, RAFT);
         return schema_builder(NAME, RAFT, std::optional(id))
-            .with_column("group_id", long_type, column_kind::partition_key)
+            .with_column("group_id", timeuuid_type, column_kind::partition_key)
             // raft log part
             .with_column("index", long_type, column_kind::clustering_key)
             .with_column("term", long_type)
@@ -246,7 +246,7 @@ schema_ptr raft_snapshots() {
     static thread_local auto schema = [] {
         auto id = generate_legacy_id(NAME, RAFT_SNAPSHOTS);
         return schema_builder(NAME, RAFT_SNAPSHOTS, std::optional(id))
-            .with_column("group_id", long_type, column_kind::partition_key)
+            .with_column("group_id", timeuuid_type, column_kind::partition_key)
             .with_column("id", uuid_type, column_kind::clustering_key)
             .with_column("idx", long_type)
             .with_column("term", long_type)
diff --git a/service/raft/raft_sys_table_storage.cc b/service/raft/raft_sys_table_storage.cc
index 18e3de63f8..4ea44afaa8 100644
--- a/service/raft/raft_sys_table_storage.cc
+++ b/service/raft/raft_sys_table_storage.cc
@@ -171,7 +171,7 @@ future<> raft_sys_table_storage::do_store_log_entries(const std::vector<raft::lo
 
         // don't include serialized "data" here since it will require to linearize the stream
         std::vector<cql3::raw_value> single_stmt_values = {
-            cql3::raw_value::make_value(long_type->decompose(_group_id.id)),
+            cql3::raw_value::make_value(timeuuid_type->decompose(_group_id.id)),
             cql3::raw_value::make_value(long_type->decompose(int64_t(eptr->term))),
             cql3::raw_value::make_value(long_type->decompose(int64_t(eptr->idx)))
         };
diff --git a/service/raft/schema_raft_state_machine.cc b/service/raft/schema_raft_state_machine.cc
index 09cd842922..47a69120c6 100644
--- a/service/raft/schema_raft_state_machine.cc
+++ b/service/raft/schema_raft_state_machine.cc
@@ -20,6 +20,8 @@
  */
 #include "service/raft/schema_raft_state_machine.hh"
 
+raft::group_id schema_raft_state_machine::gid = raft::group_id{utils::UUID_gen::min_time_UUID()};
+
 future<> schema_raft_state_machine::apply(std::vector<raft::command_cref> command) {
     return make_ready_future<>();
 }
diff --git a/service/raft/schema_raft_state_machine.hh b/service/raft/schema_raft_state_machine.hh
index bd65f378a8..b8e5edf7f6 100644
--- a/service/raft/schema_raft_state_machine.hh
+++ b/service/raft/schema_raft_state_machine.hh
@@ -21,6 +21,7 @@
 #pragma once
 
 #include "raft/raft.hh"
+#include "utils/UUID_gen.hh"
 
 // Raft state machine implementation for managing schema changes.
 // NOTE: schema raft server is always instantiated on shard 0.
@@ -28,7 +29,7 @@ class schema_raft_state_machine : public raft::state_machine {
 public:
     // schema raft server always belongs to a special pre-defined raft group with id = UUID(0, 0)
     // TODO: 0-raft group id should be unique for each scylla cluster
-    static constexpr raft::group_id gid;
+    static raft::group_id gid;
 
     future<> apply(std::vector<raft::command_cref> command) override;
     future<raft::snapshot_id> take_snapshot() override;
diff --git a/test/boost/raft_sys_table_storage_test.cc b/test/boost/raft_sys_table_storage_test.cc
index c5a602282a..1a16739dc0 100644
--- a/test/boost/raft_sys_table_storage_test.cc
+++ b/test/boost/raft_sys_table_storage_test.cc
@@ -22,7 +22,7 @@
 #include <seastar/testing/test_case.hh>
 #include <seastar/core/coroutine.hh>
 
-#include "utils/UUID.hh"
+#include "utils/UUID_gen.hh"
 
 #include "service/raft/raft_sys_table_storage.hh"
 
@@ -56,7 +56,7 @@ static bool operator==(const log_entry& lhs, const log_entry& rhs) {
 
 } // namespace raft
 
-static constexpr raft::group_id gid;
+static raft::group_id gid = raft::group_id{utils::UUID_gen::min_time_UUID()};
 
 // Create a test log with entries of each kind to test that these get
 // serialized/deserialized properly

From 18e3fcdbf1f3174c0753f5856acfc5bc8d139142 Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@scylladb.com>
Date: Thu, 27 May 2021 14:49:01 +0300
Subject: [PATCH 8/8] raft: (service) servers map is gid -> server, not sid ->
 server

Raft Group registry should map Raft Group Id to Raft Server,
not Raft Server ID (which is identical for all groups) to Raft server.

Raft Group 0 ID works as a cluster identifier, so is generated when a
new cluster is created and is shared by all nodes of the same cluster.

Implement a helper to get raft::server by group id.

Consistently throw a new raft_group_not_found exception
if there is no server or rpc for the specified group id.
---
 service/raft/raft_services.cc             | 93 ++++++++++-------------
 service/raft/raft_services.hh             | 32 +++++---
 service/raft/schema_raft_state_machine.cc |  2 -
 service/raft/schema_raft_state_machine.hh |  4 -
 test/boost/raft_sys_table_storage_test.cc |  2 +-
 5 files changed, 65 insertions(+), 68 deletions(-)

diff --git a/service/raft/raft_services.cc b/service/raft/raft_services.cc
index 1ab93d9dc0..f333f2dec4 100644
--- a/service/raft/raft_services.cc
+++ b/service/raft/raft_services.cc
@@ -24,7 +24,6 @@
 #include "service/raft/schema_raft_state_machine.hh"
 #include "service/raft/raft_gossip_failure_detector.hh"
 
-#include "raft/raft.hh"
 #include "message/messaging_service.hh"
 #include "cql3/query_processor.hh"
 #include "gms/gossiper.hh"
@@ -48,10 +47,10 @@ void raft_services::init_rpc_verbs() {
             const rpc::client_info& cinfo,
             const raft::group_id& gid, raft::server_id from, raft::server_id dst, auto handler) {
         return container().invoke_on(shard_for_group(gid),
-                [addr = netw::messaging_service::get_source(cinfo).addr, from, dst, handler] (raft_services& self) mutable {
+                [addr = netw::messaging_service::get_source(cinfo).addr, from, dst, gid, handler] (raft_services& self) mutable {
             // Update the address mappings for the rpc module
             // in case the sender is encountered for the first time
-            auto& rpc = self.get_rpc(dst);
+            auto& rpc = self.get_rpc(gid);
             // The address learnt from a probably unknown server should
             // eventually expire
             self.update_address_mapping(from, std::move(addr), true);
@@ -132,20 +131,7 @@ seastar::future<> raft_services::init() {
     // Once a Raft server starts, it soon times out
     // and starts an election, so RPC must be ready by
     // then to send VoteRequest messages.
-    init_rpc_verbs();
-    // schema raft server instance always resides on shard 0
-    if (this_shard_id() != 0) {
-        co_return;
-    }
-    co_await [this] () -> future<> {
-        raft::server_id id = raft::server_id::create_random_id();
-        co_await add_server(id, create_schema_server(id));
-    }().handle_exception([this] (auto ep) -> future<> {
-        // co_await is not allowed inside catch (...), use
-        // handle_exception instead.
-        co_await uninit_rpc_verbs();
-        std::rethrow_exception(ep);
-    });
+    co_return init_rpc_verbs();
 }
 
 seastar::future<> raft_services::uninit() {
@@ -154,48 +140,56 @@ seastar::future<> raft_services::uninit() {
     });
 }
 
-raft_rpc& raft_services::get_rpc(raft::server_id id) {
-    auto it = _servers.find(id);
+raft_services::server_for_group& raft_services::get_server_for_group(raft::group_id gid) {
+    auto it = _servers.find(gid);
     if (it == _servers.end()) {
-        on_internal_error(rslog, format("No raft server found with id = {}", id));
+        throw raft_group_not_found(gid);
     }
-    return *it->second.rpc;
+    return it->second;
 }
 
-raft_services::create_server_result raft_services::create_schema_server(raft::server_id id) {
-    auto rpc = std::make_unique<raft_rpc>(_ms, *this, schema_raft_state_machine::gid, id);
+raft_rpc& raft_services::get_rpc(raft::group_id gid) {
+    return get_server_for_group(gid).rpc;
+}
+
+raft::server& raft_services::get_server(raft::group_id gid) {
+    return *(get_server_for_group(gid).server);
+}
+
+raft_services::server_for_group raft_services::create_server_for_group(raft::group_id gid) {
+
+    raft::server_id my_id = raft::server_id::create_random_id();
+    auto rpc = std::make_unique<raft_rpc>(_ms, *this, gid, my_id);
+    // Keep a reference to a specific RPC class.
     auto& rpc_ref = *rpc;
-    auto storage = std::make_unique<raft_sys_table_storage>(_qp.local(), schema_raft_state_machine::gid);
+    auto storage = std::make_unique<raft_sys_table_storage>(_qp.local(), gid);
     auto state_machine = std::make_unique<schema_raft_state_machine>();
+    auto server = raft::create_server(my_id, std::move(rpc), std::move(state_machine),
+            std::move(storage), _fd, raft::server::configuration());
 
-    return std::pair(raft::create_server(id,
-            std::move(rpc),
-            std::move(state_machine),
-            std::move(storage),
-            _fd,
-            raft::server::configuration()), // use default raft server configuration
-        &rpc_ref);
+    // initialize the corresponding timer to tick the raft server instance
+    auto ticker = std::make_unique<ticker_type>([srv = server.get()] { srv->tick(); });
+
+    return server_for_group{
+        .gid = std::move(gid),
+        .server = std::move(server),
+        .ticker = std::move(ticker),
+        .rpc = rpc_ref,
+    };
 }
 
-future<> raft_services::add_server(raft::server_id id, create_server_result srv) {
-    auto srv_ref = std::ref(*srv.first);
-    // initialize the corresponding timer to tick the raft server instance
-    ticker_type t([srv_ref] {
-        srv_ref.get().tick();
-    });
-    auto [it, inserted] = _servers.emplace(std::pair(id, servers_value_type{
-        .server = std::move(srv.first),
-        .rpc = srv.second,
-        .ticker = std::move(t)
-    }));
+future<> raft_services::start_server_for_group(server_for_group new_grp) {
+    auto gid = new_grp.gid;
+    auto [it, inserted] = _servers.emplace(std::move(gid), std::move(new_grp));
+
     if (!inserted) {
-        on_internal_error(rslog, format("Attempt to add the second instance of raft server with the same id={}", id));
+        on_internal_error(rslog, format("Attempt to add the second instance of raft server with the same gid={}", gid));
     }
-    ticker_type& ticker_from_map = it->second.ticker;
+    auto& grp = it->second;
     try {
         // start the server instance prior to arming the ticker timer.
         // By the time the tick() is executed the server should already be initialized.
-        co_await srv_ref.get().start();
+        co_await grp.server->start();
     } catch (...) {
         // remove server from the map to prevent calling `abort()` on a
         // non-started instance when `raft_services::uninit` is called.
@@ -203,15 +197,11 @@ future<> raft_services::add_server(raft::server_id id, create_server_result srv)
         on_internal_error(rslog, std::current_exception());
     }
 
-    ticker_from_map.arm_periodic(tick_interval);
+    grp.ticker->arm_periodic(tick_interval);
 }
 
 unsigned raft_services::shard_for_group(const raft::group_id& gid) const {
-    if (gid == schema_raft_state_machine::gid) {
-        return 0; // schema raft server is always owned by shard 0
-    }
-    // We haven't settled yet on how to organize and manage (group_id -> shard_id) mapping
-    on_internal_error(rslog, format("Could not map raft group id {} to a corresponding shard id", gid));
+    return 0; // schema raft server is always owned by shard 0
 }
 
 gms::inet_address raft_services::get_inet_address(raft::server_id id) const {
@@ -229,3 +219,4 @@ void raft_services::update_address_mapping(raft::server_id id, gms::inet_address
 void raft_services::remove_address_mapping(raft::server_id id) {
     _srv_address_mappings.erase(id);
 }
+
diff --git a/service/raft/raft_services.hh b/service/raft/raft_services.hh
index 60b4dfbec6..0091e6848b 100644
--- a/service/raft/raft_services.hh
+++ b/service/raft/raft_services.hh
@@ -44,6 +44,13 @@ class gossiper;
 class raft_rpc;
 class raft_gossip_failure_detector;
 
+struct raft_group_not_found: public raft::error {
+    raft::group_id gid;
+    raft_group_not_found(raft::group_id gid_arg)
+            : raft::error(format("Raft group {} not found", gid_arg)), gid(gid_arg)
+    {}
+};
+
 // This class is responsible for creating, storing and accessing raft servers.
 // It also manages the raft rpc verbs initialization.
 //
@@ -55,22 +62,21 @@ public:
     // TODO: should be configurable.
     static constexpr ticker_type::duration tick_interval = std::chrono::milliseconds(100);
 private:
-    using create_server_result = std::pair<std::unique_ptr<raft::server>, raft_rpc*>;
-
     netw::messaging_service& _ms;
     gms::gossiper& _gossiper;
     sharded<cql3::query_processor>& _qp;
     // Shard-local failure detector instance shared among all raft groups
     shared_ptr<raft_gossip_failure_detector> _fd;
 
-    struct servers_value_type {
+    struct server_for_group {
+        raft::group_id gid;
         std::unique_ptr<raft::server> server;
-        raft_rpc* rpc;
-        ticker_type ticker;
+        std::unique_ptr<ticker_type> ticker;
+        raft_rpc& rpc;
     };
     // Raft servers along with the corresponding timers to tick each instance.
     // Currently ticking every 100ms.
-    std::unordered_map<raft::server_id, servers_value_type> _servers;
+    std::unordered_map<raft::group_id, server_for_group> _servers;
     // inet_address:es for remote raft servers known to us
     raft_address_map<> _srv_address_mappings;
 
@@ -78,8 +84,9 @@ private:
     seastar::future<> uninit_rpc_verbs();
     seastar::future<> stop_servers();
 
-    create_server_result create_schema_server(raft::server_id id);
+    server_for_group create_server_for_group(raft::group_id id);
 
+    server_for_group& get_server_for_group(raft::group_id id);
 public:
 
     raft_services(netw::messaging_service& ms, gms::gossiper& gs, sharded<cql3::query_processor>& qp);
@@ -97,10 +104,15 @@ public:
     // Must be invoked explicitly on each shard to stop this service.
     seastar::future<> uninit();
 
-    raft_rpc& get_rpc(raft::server_id id);
+    raft_rpc& get_rpc(raft::group_id gid);
+
+    // Find server for group by group id. Throws exception if
+    // there is no such group.
+    raft::server& get_server(raft::group_id gid);
+
     // Start raft server instance, store in the map of raft servers and
-    // initialize the associated timer to tick the server.
-    future<> add_server(raft::server_id id, create_server_result srv);
+    // arm the associated timer to tick the server.
+    future<> start_server_for_group(server_for_group grp);
     unsigned shard_for_group(const raft::group_id& gid) const;
 
     // Map raft server_id to inet_address to be consumed by `messaging_service`
diff --git a/service/raft/schema_raft_state_machine.cc b/service/raft/schema_raft_state_machine.cc
index 47a69120c6..09cd842922 100644
--- a/service/raft/schema_raft_state_machine.cc
+++ b/service/raft/schema_raft_state_machine.cc
@@ -20,8 +20,6 @@
  */
 #include "service/raft/schema_raft_state_machine.hh"
 
-raft::group_id schema_raft_state_machine::gid = raft::group_id{utils::UUID_gen::min_time_UUID()};
-
 future<> schema_raft_state_machine::apply(std::vector<raft::command_cref> command) {
     return make_ready_future<>();
 }
diff --git a/service/raft/schema_raft_state_machine.hh b/service/raft/schema_raft_state_machine.hh
index b8e5edf7f6..69204f1d65 100644
--- a/service/raft/schema_raft_state_machine.hh
+++ b/service/raft/schema_raft_state_machine.hh
@@ -27,10 +27,6 @@
 // NOTE: schema raft server is always instantiated on shard 0.
 class schema_raft_state_machine : public raft::state_machine {
 public:
-    // schema raft server always belongs to a special pre-defined raft group with id = UUID(0, 0)
-    // TODO: 0-raft group id should be unique for each scylla cluster
-    static raft::group_id gid;
-
     future<> apply(std::vector<raft::command_cref> command) override;
     future<raft::snapshot_id> take_snapshot() override;
     void drop_snapshot(raft::snapshot_id id) override;
diff --git a/test/boost/raft_sys_table_storage_test.cc b/test/boost/raft_sys_table_storage_test.cc
index 1a16739dc0..fbbd51c456 100644
--- a/test/boost/raft_sys_table_storage_test.cc
+++ b/test/boost/raft_sys_table_storage_test.cc
@@ -56,7 +56,7 @@ static bool operator==(const log_entry& lhs, const log_entry& rhs) {
 
 } // namespace raft
 
-static raft::group_id gid = raft::group_id{utils::UUID_gen::min_time_UUID()};
+static raft::group_id gid{utils::UUID_gen::min_time_UUID()};
 
 // Create a test log with entries of each kind to test that these get
 // serialized/deserialized properly