Merge pull request #743 from mmaslankaprv/exit-early-on-invariants-ch…

…ange

Exit early on invariants change

src/v/cluster/controller.cc

@@ -74,6 +74,10 @@ ss::future<> controller::start() {
             std::ref(_storage),
             std::ref(_as));
       })
+      .then([this] {
+          // validate configuration invariants to exit early
+          return _members_manager.local().validate_configuration_invariants();
+      })
       .then([this] {
           return _stm.start_single(
             std::ref(clusterlog),

src/v/cluster/members_manager.cc

@@ -55,15 +55,13 @@ members_manager::members_manager(
 
 ss::future<> members_manager::start() {
     vlog(clusterlog.info, "starting cluster::members_manager...");
-    // validate node id change
-    return validate_configuration_invariants().then([this] {
-        // join raft0
-        if (!is_already_member()) {
-            join_raft0();
-        }
 
-        return start_config_changes_watcher();
-    });
+    // join raft0
+    if (!is_already_member()) {
+        join_raft0();
+    }
+
+    return start_config_changes_watcher();
 }
 
 ss::future<> members_manager::start_config_changes_watcher() {

src/v/cluster/members_manager.h

@@ -44,7 +44,7 @@ class members_manager {
 
     ss::future<> start();
     ss::future<> stop();
-
+    ss::future<> validate_configuration_invariants();
     ss::future<result<join_reply>> handle_join_request(model::broker);
     ss::future<std::error_code> apply_update(model::record_batch);
 
@@ -75,7 +75,7 @@ class members_manager {
     // Raft 0 config updates
     ss::future<> handle_raft0_cfg_update(raft::group_configuration);
     ss::future<> update_connections(patch<broker_ptr>);
-    ss::future<> validate_configuration_invariants();
+
     ss::future<> start_config_changes_watcher();
 
     ss::future<> maybe_update_current_node_configuration();

src/v/cluster/partition_allocator.cc

@@ -182,8 +182,12 @@ void partition_allocator::update_allocation_state(
       [](const model::broker_shard& l, const model::broker_shard& r) {
           return l.node_id > r.node_id;
       });
-
-    auto it = find_node(std::cbegin(shards)->node_id);
+    auto node_id = std::cbegin(shards)->node_id;
+    auto it = find_node(node_id);
+    vassert(
+      it != _machines.end(),
+      "node: {} must exists in partition allocator, it currenlty does not",
+      node_id);
     for (auto const& bs : shards) {
         // Thanks to shards being sorted we need to do only
         //  as many lookups as there are brokers