Fix[MQB]: use CSL to update state on QueueAssignmentAdvisory

src/groups/mqb/mqbblp/mqbblp_clusterstatemanager.cpp

@@ -70,19 +70,24 @@ void ClusterStateManager::onCommit(
     BSLS_ASSERT_SAFE(dispatcher()->inDispatcherThread(d_cluster_p));
     BSLS_ASSERT_SAFE(advisory.choice().isClusterMessageValue());
 
-    // NOTE: Even when using old workflow, we still apply all advisories to the
-    // CSL. We just don't invoke the commit callbacks.
-    if (!d_clusterConfig.clusterAttributes().isCSLModeEnabled()) {
-        return;  // RETURN
-    }
-
     if (status != mqbc::ClusterStateLedgerCommitStatus::e_SUCCESS) {
         BALL_LOG_ERROR << d_clusterData_p->identity().description()
                        << ": Failed to commit advisory: " << advisory
                        << ", with status '" << status << "'";
         return;  // RETURN
     }
 
+    const bmqp_ctrlmsg::ClusterMessage& clusterMessage =
+        advisory.choice().clusterMessage();
+
+    // NOTE: Even when using old workflow, we still apply all advisories to the
+    // CSL. We just don't invoke the commit callbacks.
+    // Make an exception for QueueAssignmentAdvisory
+    if (!d_clusterConfig.clusterAttributes().isCSLModeEnabled() &&
+        !clusterMessage.choice().isQueueAssignmentAdvisoryValue()) {
+        return;  // RETURN
+    }
+
     // Commenting out following 'if' check to fix an assert during node
     // shutdown.
     // if (   d_clusterData_p->membership().selfNodeStatus()
@@ -94,8 +99,6 @@ void ClusterStateManager::onCommit(
                   << ": Committed advisory: " << advisory << ", with status '"
                   << status << "'";
 
-    const bmqp_ctrlmsg::ClusterMessage& clusterMessage =
-        advisory.choice().clusterMessage();
     mqbc::ClusterUtil::apply(d_state_p, clusterMessage, *d_clusterData_p);
 }
 
@@ -1690,16 +1693,21 @@ void ClusterStateManager::processQueueAssignmentAdvisory(
             }
         }
         else {
-            AppInfos appIdInfos(d_allocator_p);
+            if (delayed) {
+                AppInfos appIdInfos(d_allocator_p);
 
-            mqbc::ClusterUtil::parseQueueInfo(&appIdInfos,
-                                              queueInfo,
-                                              d_allocator_p);
+                mqbc::ClusterUtil::parseQueueInfo(&appIdInfos,
+                                                  queueInfo,
+                                                  d_allocator_p);
 
-            d_state_p->assignQueue(uri,
-                                   queueKey,
-                                   queueInfo.partitionId(),
-                                   appIdInfos);
+                d_state_p->assignQueue(uri,
+                                       queueKey,
+                                       queueInfo.partitionId(),
+                                       appIdInfos);
+            }
+            // When this function is not buffered, called from
+            // processQueueAssignmentAdvisory, assignQueue will
+            // be triggered through mqbblp::ClusterStateManager::onCommit
         }
 
         BALL_LOG_INFO << d_cluster_p->description()
@@ -2150,7 +2158,9 @@ void ClusterStateManager::processLeaderAdvisory(
     queueAsgnAdv.sequenceNumber() = advisory.sequenceNumber();
     queueAsgnAdv.queues()         = advisory.queues();
 
-    processQueueAssignmentAdvisory(controlMsg, source);
+    processQueueAssignmentAdvisory(controlMsg,
+                                   source,
+                                   false /* not delayed */);
 
     // Leader status and sequence number are updated unconditionally.  It may
     // have been updated by one of the routines called earlier in this method,

src/groups/mqb/mqbc/mqbc_clusterutil.cpp

@@ -1009,34 +1009,6 @@ ClusterUtil::assignQueue(ClusterState*         clusterState,
                        << queueAdvisory << ", rc: " << rc;
     }
 
-    if (!cluster->isCSLModeEnabled()) {
-        // In CSL mode, we assign the queue to ClusterState upon CSL commit
-        // callback of QueueAssignmentAdvisory, so we don't assign it here.
-
-        // In non-CSL mode this is the shortcut to call Primary CQH instead of
-        // waiting for the quorum of acks in the ledger.
-
-        BSLS_ASSERT_SAFE(queueAdvisory.queues().size() == 1);
-
-        bmqp_ctrlmsg::QueueInfo& queueInfo = queueAdvisory.queues().back();
-
-        AppInfos appInfos(allocator);
-        mqbc::ClusterUtil::parseQueueInfo(&appInfos, queueInfo, allocator);
-
-        BSLA_MAYBE_UNUSED const bool assignRc = clusterState->assignQueue(
-            uri,
-            key,
-            queueAdvisory.queues().back().partitionId(),
-            appInfos);
-        BSLS_ASSERT_SAFE(assignRc);
-
-        BALL_LOG_INFO << cluster->description()
-                      << ": Queue assigned: " << queueAdvisory;
-
-        // Broadcast 'queueAssignmentAdvisory' to all followers
-        clusterData->messageTransmitter().broadcastMessage(controlMsg);
-    }
-
     return QueueAssignmentResult::k_ASSIGNMENT_OK;
 }
 

src/groups/mqb/mqbi/mqbi_clusterstatemanager.h

@@ -86,9 +86,9 @@ class ClusterStateManager {
         AfterPartitionPrimaryAssignmentCb;
 
     /// Pair of (appId, appKey)
-    typedef bsl::pair<bsl::string, mqbu::StorageKey> AppInfo;
+    typedef bsl::pair<bsl::string, mqbu::StorageKey>          AppInfo;
     typedef bsl::unordered_map<bsl::string, mqbu::StorageKey> AppInfos;
-    typedef AppInfos::const_iterator                 AppInfosCIter;
+    typedef AppInfos::const_iterator                          AppInfosCIter;
 
     struct QueueAssignmentResult {
         enum Enum {

src/integration-tests/test_migrate_to_csl.py

@@ -0,0 +1,54 @@
+# Copyright 2025 Bloomberg Finance L.P.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Testing migrating to CSL.
+"""
+import blazingmq.dev.it.testconstants as tc
+from blazingmq.dev.it.fixtures import (  # pylint: disable=unused-import
+    Cluster,
+    multi_node,
+)
+
+
+def test_assign_queue(multi_node: Cluster):
+    """
+    This test is to make sure when a queue assignment happens,
+    it goes thought the CSL path, not the non-CSL.
+    """
+    proxies = multi_node.proxy_cycle()
+    proxy = next(proxies)
+
+    leader = multi_node.last_known_leader
+    members = multi_node.nodes(exclude=leader)
+
+    uri = tc.URI_PRIORITY_SC
+
+    producer = proxy.create_client("producer")
+    producer.open(uri, flags=["write"], succeed=True)
+
+    timeout = 1
+
+    for member in members:
+        assert member.outputs_regex(
+            "(Applying cluster message with type = UPDATE).*(queueAssignmentAdvisory)",
+            timeout,
+        )
+        assert member.outputs_regex(
+            "(Committed advisory).*queueAssignmentAdvisory", timeout
+        )
+        assert not member.outputs_regex(
+            "'QueueUnAssignmentAdvisory' will be applied to", timeout
+        )