From 010561e05369b2e6a282bbafc0688ef6ab48aaea Mon Sep 17 00:00:00 2001
From: Michal Maslanka <michal@redpanda.com>
Date: Thu, 5 Oct 2023 16:13:11 +0200
Subject: [PATCH] c/config_manager: populate configuration status on members
 notification

When member is added to the cluster `cluster::configuration_manager`
should be immediately aware of it instead of waiting for the first
status update.

Wired up cluster member added notification to update cluster
configuration reconciliation status with initial state for joining
member.

Fixes: #13497
Fixes: #13503

Signed-off-by: Michal Maslanka <michal@redpanda.com>
---
 src/v/cluster/config_manager.cc | 13 ++++++++-----
 src/v/cluster/config_manager.h  |  2 +-
 2 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/src/v/cluster/config_manager.cc b/src/v/cluster/config_manager.cc
index e24915be744f6..c321a78908cfc 100644
--- a/src/v/cluster/config_manager.cc
+++ b/src/v/cluster/config_manager.cc
@@ -21,6 +21,7 @@
 #include "config/configuration.h"
 #include "config/node_config.h"
 #include "features/feature_table.h"
+#include "model/metadata.h"
 #include "resource_mgmt/io_priority.h"
 #include "rpc/connection_cache.h"
 #include "utils/file_io.h"
@@ -81,7 +82,7 @@ config_manager::config_manager(
         /**
          * Register notification immediately not to lose status updates.
          */
-        _member_removed_notification
+        _member_update_notification
           = _members.local().register_members_updated_notification(
             [this](model::node_id id, model::membership_state new_state) {
                 handle_cluster_members_update(id, new_state);
@@ -229,17 +230,19 @@ ss::future<> config_manager::start() {
 }
 void config_manager::handle_cluster_members_update(
   model::node_id id, model::membership_state new_state) {
-    if (new_state != model::membership_state::removed) {
-        return;
+    if (new_state == model::membership_state::active) {
+        // add an empty status placeholder if node is not yet known
+        status.try_emplace(id);
+    } else if (new_state != model::membership_state::removed) {
+        status.erase(id);
     }
-    status.erase(id);
 }
 
 ss::future<> config_manager::stop() {
     vlog(clusterlog.info, "Stopping Config Manager...");
     _reconcile_wait.broken();
     _members.local().unregister_members_updated_notification(
-      _member_removed_notification);
+      _member_update_notification);
     _leaders.local().unregister_leadership_change_notification(
       _raft0_leader_changed_notification);
     co_await _gate.close();
diff --git a/src/v/cluster/config_manager.h b/src/v/cluster/config_manager.h
index bd3e53e1f445e..3554a4b35f2e3 100644
--- a/src/v/cluster/config_manager.h
+++ b/src/v/cluster/config_manager.h
@@ -135,7 +135,7 @@ class config_manager final {
     ss::sharded<partition_leaders_table>& _leaders;
     ss::sharded<features::feature_table>& _feature_table;
     ss::sharded<cluster::members_table>& _members;
-    notification_id_type _member_removed_notification;
+    notification_id_type _member_update_notification;
     notification_id_type _raft0_leader_changed_notification;
 
     ss::condition_variable _reconcile_wait;