mqbc: Construct merged state with uncommitted CSLs before send

kaikulimu · kaikulimu · commit aa6e331ed7ac · 2025-11-07T16:56:40.000-05:00
Signed-off-by: Yuan Jing Vincent Yan &lt;yyan82@bloomberg.net&gt;
diff --git a/src/groups/mqb/mqbblp/mqbblp_clusterorchestrator.cpp b/src/groups/mqb/mqbblp/mqbblp_clusterorchestrator.cpp
@@ -1084,14 +1084,10 @@ void ClusterOrchestrator::processNodeStatusAdvisory(
                 // thread scheduling, 'source' may see that heartbeat after the
                 // first advisory.
 
-                bsl::vector<bmqp_ctrlmsg::PartitionPrimaryInfo> partitions;
-                mqbc::ClusterUtil::loadPartitionsInfo(&partitions,
-                                                      *clusterState());
                 d_stateManager_mp->sendClusterState(
                     true,  // sendPartitionPrimaryInfo
                     true,  // sendQueuesInfo
-                    source,
-                    partitions);
+                    source);
             }
         }
         else if (d_clusterConfig.clusterAttributes().isFSMWorkflow() &&
@@ -1201,14 +1197,10 @@ void ClusterOrchestrator::processNodeStateChangeEvent(
                     << ": leader (self) is ACTIVE; will send leader"
                     << " advisory to new node: " << node->nodeDescription();
 
-                bsl::vector<bmqp_ctrlmsg::PartitionPrimaryInfo> partitions;
-                mqbc::ClusterUtil::loadPartitionsInfo(&partitions,
-                                                      *clusterState());
                 d_stateManager_mp->sendClusterState(
                     true,  // sendPartitionPrimaryInfo
                     true,  // sendQueuesInfo
-                    node,
-                    partitions);
+                    node);
             }
         }
 
@@ -1827,12 +1819,9 @@ void ClusterOrchestrator::processLeaderPassiveNotification(
     }
 
     // Self is an ACTIVE leader - broadcast 'LeaaderAdvisory'
-    bsl::vector<bmqp_ctrlmsg::PartitionPrimaryInfo> partitions;
-    mqbc::ClusterUtil::loadPartitionsInfo(&partitions, *clusterState());
     d_stateManager_mp->sendClusterState(true,  // sendPartitionPrimaryInfo
                                         true,  // sendQueuesInfo
-                                        0,
-                                        partitions);
+                                        0);
 }
 
 void ClusterOrchestrator::onRecoverySuccess()
diff --git a/src/groups/mqb/mqbblp/mqbblp_clusterstatemanager.cpp b/src/groups/mqb/mqbblp/mqbblp_clusterstatemanager.cpp
@@ -911,6 +911,7 @@ void ClusterStateManager::sendClusterState(
                                         *d_state_p,
                                         sendPartitionPrimaryInfo,
                                         sendQueuesInfo,
+                                        d_allocator_p,
                                         node,
                                         partitions);
 }
diff --git a/src/groups/mqb/mqbc/mqbc_clusterstatemanager.cpp b/src/groups/mqb/mqbc/mqbc_clusterstatemanager.cpp
@@ -1542,6 +1542,7 @@ void ClusterStateManager::sendClusterState(
                                         *d_state_p,
                                         sendPartitionPrimaryInfo,
                                         sendQueuesInfo,
+                                        d_allocator_p,
                                         node,
                                         partitions);
 }
diff --git a/src/groups/mqb/mqbc/mqbc_clusterutil.cpp b/src/groups/mqb/mqbc/mqbc_clusterutil.cpp
@@ -14,6 +14,7 @@
 // limitations under the License.
 
 // mqbc_clusterutil.cpp                                               -*-C++-*-
+#include "mqbc_clusterstateledger.h"
 #include <mqbc_clusterutil.h>
 
 #include <mqbscm_version.h>
@@ -1394,8 +1395,9 @@ void ClusterUtil::sendClusterState(
     const ClusterState&  clusterState,
     bool                 sendPartitionPrimaryInfo,
     bool                 sendQueuesInfo,
+    bslma::Allocator*    allocator,
     mqbnet::ClusterNode* node,
-    const bsl::vector<bmqp_ctrlmsg::PartitionPrimaryInfo>& partitions)
+    const bsl::vector<bmqp_ctrlmsg::PartitionPrimaryInfo>& newlyAssigned)
 {
     // executed by the cluster *DISPATCHER* thread
 
@@ -1406,6 +1408,13 @@ void ClusterUtil::sendClusterState(
     BSLS_ASSERT_SAFE(mqbnet::ElectorState::e_LEADER ==
                      clusterData->electorInfo().electorState());
     BSLS_ASSERT_SAFE(ledger && ledger->isOpen());
+    if (!sendPartitionPrimaryInfo) {
+        BSLS_ASSERT_SAFE(newlyAssigned.empty());
+    }
+    else if (!newlyAssigned.empty()) {
+        BSLS_ASSERT_SAFE(newlyAssigned.size() ==
+                         clusterState.partitions().size());
+    }
 
     if (clusterData->clusterConfig().clusterAttributes().isFSMWorkflow()) {
         // In FSM mode, the *only* possible caller of this method is
@@ -1414,14 +1423,6 @@ void ClusterUtil::sendClusterState(
         BSLS_ASSERT_SAFE(sendPartitionPrimaryInfo && !sendQueuesInfo);
     }
 
-    if (sendPartitionPrimaryInfo) {
-        BSLS_ASSERT_SAFE(partitions.size() ==
-                         clusterState.partitions().size());
-    }
-    else {
-        BSLS_ASSERT_SAFE(partitions.empty());
-    };
-
     if (bmqp_ctrlmsg::NodeStatus::E_STOPPING ==
         clusterData->membership().selfNodeStatus()) {
         // No need to send cluster state since self is stopping.  After self
@@ -1445,8 +1446,47 @@ void ClusterUtil::sendClusterState(
         clusterData->electorInfo().nextLeaderMessageSequence(
             &advisory.sequenceNumber());
 
-        advisory.partitions() = partitions;
-        loadQueuesInfo(&advisory.queues(), clusterState);
+        if (!newlyAssigned.empty()) {
+            // If we are sending a cluster state snapshot and there are new
+            // assignments, it implies that this will be the first leader
+            // advisory and thus there won't be any uncommitted CSL advisories.
+            advisory.partitions() = newlyAssigned;
+            loadQueuesInfo(&advisory.queues(), clusterState);
+        }
+        else {
+            // We construct a "merged state" of current cluster state and
+            // information from uncommitted CSL advisories, before sending over
+            // to follower(s).
+            //
+            // NOTE: This code path is *only* reachable in non-FSM mode.
+            ClusterState tempState(
+                &clusterData->cluster(),
+                clusterData->clusterConfig().partitionConfig().numPartitions(),
+                allocator);
+            apply(&tempState, clusterMessage, *clusterData);
+            const int rc = load(&tempState,
+                                ledger->getIterator().get(),
+                                *clusterData,
+                                allocator);
+            if (rc != 0) {
+                BALL_LOG_ERROR
+                    << clusterData->identity().description()
+                    << ": Failed to load CSL content to temp cluster "
+                    << "state as part of 'sendClusterState', rc: " << rc;
+            }
+
+            ClusterStateLedger::ClusterMessageCRefList uncommittedAdvisories;
+            ledger->uncommittedAdvisories(&uncommittedAdvisories);
+            for (ClusterStateLedger::ClusterMessageCRefList::const_iterator
+                     cit = uncommittedAdvisories.begin();
+                 cit != uncommittedAdvisories.end();
+                 ++cit) {
+                apply(&tempState, *cit, *clusterData);
+            }
+
+            loadPartitionsInfo(&advisory.partitions(), tempState);
+            loadQueuesInfo(&advisory.queues(), tempState);
+        }
     }
     else if (sendPartitionPrimaryInfo) {
         bmqp_ctrlmsg::PartitionPrimaryAdvisory& advisory =
@@ -1455,7 +1495,7 @@ void ClusterUtil::sendClusterState(
         clusterData->electorInfo().nextLeaderMessageSequence(
             &advisory.sequenceNumber());
 
-        advisory.partitions() = partitions;
+        advisory.partitions() = newlyAssigned;
     }
     else {
         BSLS_ASSERT_SAFE(sendQueuesInfo);
@@ -1960,7 +2000,7 @@ void ClusterUtil::loadPartitionsInfo(
     const ClusterState&                              state)
 {
     // PRECONDITIONS
-    BSLS_ASSERT_SAFE(out);
+    BSLS_ASSERT_SAFE(out && out->empty());
 
     for (int pid = 0; pid < static_cast<int>(state.partitions().size());
          ++pid) {
diff --git a/src/groups/mqb/mqbc/mqbc_clusterutil.h b/src/groups/mqb/mqbc/mqbc_clusterutil.h
@@ -272,14 +272,15 @@ struct ClusterUtil {
                  bslma::Allocator*               allocator);
 
     /// Send the current cluster state to follower nodes.  If the specified
-    /// `sendPartitionPrimaryInfo` is true, the specified partition-primary
-    /// mapping `partitions` will be included.  If the specified
-    /// `sendQueuesInfo` is true, queue-partition assignments will be
+    /// `sendPartitionPrimaryInfo` is true, the partition-primary assignments
+    /// will be included.  The optionally specified `newlyAssigned` will be
+    /// sent instead of the current assignments.  If the specified
+    /// `sendQueuesInfo` is true, the queue-partition assignments will be
     /// included.  If the optionally specified `node` is non-null, send the
     /// cluster state to that `node` only.  Otherwise, broadcast to all
-    /// followers.  Behavior is undefined unless this node is the leader,
-    /// and at least one of `sendPartitionPrimaryInfo` or `sendQueuesInfo`
-    /// is true.
+    /// followers.  Use the specified `allocator` for memory allocations.
+    /// Behavior is undefined unless this node is the leader, and at least one
+    /// of `sendPartitionPrimaryInfo` or `sendQueuesInfo` is true.
     ///
     /// THREAD: This method is invoked in the associated cluster's
     ///         dispatcher thread.
@@ -289,8 +290,9 @@ struct ClusterUtil {
         const ClusterState&  clusterState,
         bool                 sendPartitionPrimaryInfo,
         bool                 sendQueuesInfo,
+        bslma::Allocator*    allocator,
         mqbnet::ClusterNode* node = 0,
-        const bsl::vector<bmqp_ctrlmsg::PartitionPrimaryInfo>& partitions =
+        const bsl::vector<bmqp_ctrlmsg::PartitionPrimaryInfo>& newlyAssigned =
             bsl::vector<bmqp_ctrlmsg::PartitionPrimaryInfo>());
 
     /// Append to the specified `out` a newly created cluster node

Original file line number	Diff line number	Diff line change
`@@ -911,6 +911,7 @@ void ClusterStateManager::sendClusterState(`
`911`	`911`	`*d_state_p,`
`912`	`912`	`sendPartitionPrimaryInfo,`
`913`	`913`	`sendQueuesInfo,`
	`914`	`+ d_allocator_p,`
`914`	`915`	`node,`
`915`	`916`	`partitions);`
`916`	`917`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1542,6 +1542,7 @@ void ClusterStateManager::sendClusterState(`
`1542`	`1542`	`*d_state_p,`
`1543`	`1543`	`sendPartitionPrimaryInfo,`
`1544`	`1544`	`sendQueuesInfo,`
	`1545`	`+ d_allocator_p,`
`1545`	`1546`	`node,`
`1546`	`1547`	`partitions);`
`1547`	`1548`	`}`