Fix handle csl failure to assign queue (#821)

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

Author: dorjesinpo

src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp

@@ -89,8 +89,7 @@ namespace BloombergLP {
 namespace mqbblp {
 
 namespace {
-const char k_MAXIMUM_NUMBER_OF_QUEUES_REACHED[] =
-    "maximum number of queues reached";
+
 const char k_SELF_NODE_IS_STOPPING[] = "self node is stopping";
 
 const int k_MAX_INSTANT_MESSAGES = 10;
@@ -393,8 +392,7 @@ void ClusterQueueHelper::afterPartitionPrimaryAssignment(
     }
 }
 
-mqbi::ClusterStateManager::QueueAssignmentResult::Enum
-ClusterQueueHelper::assignQueue(const QueueContextSp& queueContext)
+void ClusterQueueHelper::assignQueue(const QueueContextSp& queueContext)
 {
     // executed by the cluster *DISPATCHER* thread
 
@@ -407,30 +405,31 @@ ClusterQueueHelper::assignQueue(const QueueContextSp& queueContext)
         // Assigning a queue in a remote, is simply giving it a new queueId.
         queueContext->d_liveQInfo.d_id = getNextQueueId();
         onQueueContextAssigned(queueContext);
-        return QueueAssignmentResult::k_ASSIGNMENT_OK;  // RETURN
     }
-
-    if (d_clusterData_p->electorInfo().hasActiveLeader()) {
+    else if (d_clusterData_p->electorInfo().hasActiveLeader()) {
         if (d_clusterData_p->electorInfo().isSelfLeader()) {
-            return d_clusterStateManager_p->assignQueue(queueContext->uri());
-            // RETURN
+            bmqp_ctrlmsg::Status status(d_allocator_p);
+
+            if (!d_clusterStateManager_p->assignQueue(queueContext->uri(),
+                                                      &status)) {
+                processRejectedQueueAssignment(queueContext.get(), status);
+            }
+            // else, all other failure are transient. will retry.
         }
         else {
             requestQueueAssignment(queueContext->uri());
-            return QueueAssignmentResult::k_ASSIGNMENT_OK;  // RETURN
         }
     }
+    else {
+        // Queue not yet assigned, because we don't have a leader (or leader is
+        // not active) at the moment, nothing to be done; the queue will
+        // automatically be re-processed once we have an active leader.
 
-    // Queue not yet assigned, because we don't have a leader (or leader is not
-    // active) at the moment, nothing to be done; the queue will automatically
-    // be re-processed once we have an active leader.
-
-    BALL_LOG_INFO << d_cluster_p->description()
-                  << " Cannot proceed with queueAssignment of "
-                  << "'" << queueContext->uri()
-                  << "' (waiting for an ACTIVE leader).";
-
-    return QueueAssignmentResult::k_ASSIGNMENT_OK;
+        BALL_LOG_INFO << d_cluster_p->description()
+                      << " Cannot proceed with queueAssignment of '"
+                      << queueContext->uri()
+                      << "' (waiting for an ACTIVE leader).";
+    }
 }
 
 void ClusterQueueHelper::requestQueueAssignment(const bmqt::Uri& uri)
@@ -564,6 +563,7 @@ void ClusterQueueHelper::onQueueAssignmentResponse(
         // exists and the queue is assigned, because the
         // 'queueAssignmentAdvisory' message may have been dropped due to
         // change of leader.
+
         BALL_LOG_INFO << d_cluster_p->description()
                       << " Received queueAssignment response from '"
                       << responder->nodeDescription()
@@ -589,24 +589,22 @@ void ClusterQueueHelper::onQueueAssignmentResponse(
             // time.
         }
         else if (requestContext->result() == bmqt::GenericResult::e_REFUSED) {
-            if (requestContext->response().choice().status().code() ==
-                mqbi::ClusterErrorCode::e_NOT_LEADER) {
+            if (status.code() == mqbi::ClusterErrorCode::e_NOT_LEADER) {
                 // The leader changed by the time our request reached it; we
                 // don't have to do anything here: since the leader changed, we
                 // must have (or will shortly) received a notification about
                 // the new leader, and in the 'onClusterLeader' one thing we do
                 // is re-emit an assignmentRequest for any unassigned queue,
                 // this current one being part of them.
             }
-            else if (requestContext->response().choice().status().code() ==
-                     mqbi::ClusterErrorCode::e_LIMIT) {
+            else if (status.code() == mqbi::ClusterErrorCode::e_LIMIT ||
+                     status.code() == mqbi::ClusterErrorCode::e_CSL_FAILURE ||
+                     status.code() == mqbi::ClusterErrorCode::e_UNKNOWN) {
                 QueueContextMapIter qit = d_queues.find(uri);
                 BSLS_ASSERT_SAFE(qit != d_queues.end());
-                bdlma::LocalSequentialAllocator<256> localAllocator(
-                    d_allocator_p);
-                bsl::vector<QueueContext*> rejected(1, &localAllocator);
-                *rejected.begin() = qit->second.get();
-                processRejectedQueueAssignments(rejected);
+                const QueueContext* rejected = qit->second.get();
+
+                processRejectedQueueAssignment(rejected, status);
             }
         }
         else {
@@ -3399,35 +3397,27 @@ void ClusterQueueHelper::restoreState(int partitionId)
     }
 }
 
-void ClusterQueueHelper::processRejectedQueueAssignments(
-    const bsl::vector<QueueContext*>& rejected)
+void ClusterQueueHelper::processRejectedQueueAssignment(
+    const QueueContext*         rejected,
+    const bmqp_ctrlmsg::Status& status)
 {
     // executed by the cluster *DISPATCHER* thread
 
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(
         d_cluster_p->dispatcher()->inDispatcherThread(d_cluster_p));
 
-    bmqp_ctrlmsg::Status failure;
-    failure.category() = bmqp_ctrlmsg::StatusCategory::E_REFUSED;
-    failure.code()     = mqbi::ClusterErrorCode::e_LIMIT;
-    failure.message()  = k_MAXIMUM_NUMBER_OF_QUEUES_REACHED;
-
-    for (bsl::vector<QueueContext*>::const_iterator sIt = rejected.begin();
-         sIt != rejected.end();
-         ++sIt) {
-        for (bsl::vector<OpenQueueContextSp>::iterator
-                 cIt   = (*sIt)->d_liveQInfo.d_pending.begin(),
-                 cLast = (*sIt)->d_liveQInfo.d_pending.end();
-             cIt != cLast;
-             ++cIt) {
-            (*cIt)->d_callback(failure,
-                               0,
-                               bmqp_ctrlmsg::OpenQueueResponse(),
-                               mqbi::Cluster::OpenQueueConfirmationCookie());
-        }
-        d_queues.erase((*sIt)->uri());
-    }
+    for (bsl::vector<OpenQueueContextSp>::const_iterator
+             cIt   = rejected->d_liveQInfo.d_pending.begin(),
+             cLast = rejected->d_liveQInfo.d_pending.end();
+         cIt != cLast;
+         ++cIt) {
+        (*cIt)->d_callback(status,
+                           0,
+                           bmqp_ctrlmsg::OpenQueueResponse(),
+                           mqbi::Cluster::OpenQueueConfirmationCookie());
+    }
+    d_queues.erase(rejected->uri());
 }
 
 void ClusterQueueHelper::restoreStateRemote()
@@ -3450,9 +3440,6 @@ void ClusterQueueHelper::restoreStateRemote()
     }
 
     // Attempt to re-issue open-queue requests for all applicable queues.
-    bdlma::LocalSequentialAllocator<1024> localAllocator(d_allocator_p);
-    bsl::vector<QueueContext*>            rejected(&localAllocator);
-    rejected.reserve(d_queues.size());
 
     for (QueueContextMapConstIter cit = d_queues.cbegin();
          cit != d_queues.cend();
@@ -3473,10 +3460,9 @@ void ClusterQueueHelper::restoreStateRemote()
 
         if (!isQueueAssigned(*queueContext.get())) {
             // Queue is not assigned to a partition; get it assigned.
-            if (QueueAssignmentResult::k_ASSIGNMENT_REJECTED ==
-                assignQueue(queueContext)) {
-                rejected.push_back(queueContext.get());
-            }
+
+            assignQueue(queueContext);
+
             continue;  // CONTINUE
         }
 
@@ -3508,8 +3494,6 @@ void ClusterQueueHelper::restoreStateRemote()
 
         onQueueContextAssigned(queueContext);
     }
-
-    processRejectedQueueAssignments(rejected);
 }
 
 void ClusterQueueHelper::restoreStateCluster(int partitionId)
@@ -3595,10 +3579,6 @@ void ClusterQueueHelper::restoreStateCluster(int partitionId)
                         d_clusterData_p->membership().selfNode();
     }
 
-    bdlma::LocalSequentialAllocator<1024> localAllocator(d_allocator_p);
-    bsl::vector<QueueContext*>            rejected(&localAllocator);
-    rejected.reserve(d_queues.size());
-
     for (QueueContextMapConstIter cit = d_queues.cbegin();
          cit != d_queues.cend();
          ++cit) {
@@ -3626,10 +3606,9 @@ void ClusterQueueHelper::restoreStateCluster(int partitionId)
                 // Queue is not assigned to a partition; get it assigned.  If
                 // self is leader, it will assign it locally, if not it will
                 // send a request to the leader, etc.
-                if (QueueAssignmentResult::k_ASSIGNMENT_REJECTED ==
-                    assignQueue(queueContext)) {
-                    rejected.push_back(queueContext.get());
-                }
+
+                assignQueue(queueContext);
+
                 continue;  // CONTINUE
             }
         }
@@ -3757,8 +3736,6 @@ void ClusterQueueHelper::restoreStateCluster(int partitionId)
             }
         }
     }
-
-    processRejectedQueueAssignments(rejected);
 }
 
 bmqt::GenericResult::Enum
@@ -4699,14 +4676,7 @@ void ClusterQueueHelper::openQueue(
         queueContext->d_liveQInfo.d_pending.push_back(context);
 
         // Initiate the assignment.
-        if (QueueAssignmentResult::k_ASSIGNMENT_REJECTED ==
-            assignQueue(queueContext)) {
-            bdlma::LocalSequentialAllocator<1024> localAllocator(
-                d_allocator_p);
-            bsl::vector<QueueContext*> rejected(&localAllocator);
-            rejected.push_back(queueContext.get());
-            processRejectedQueueAssignments(rejected);
-        }
+        assignQueue(queueContext);
     }
 }
 

src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.h

@@ -292,9 +292,6 @@ class ClusterQueueHelper BSLS_KEYWORD_FINAL
     typedef mqbc::ClusterState::DomainStates           DomainStates;
     typedef mqbc::ClusterState::DomainStatesCIter      DomainStatesCIter;
 
-    typedef mqbi::ClusterStateManager::QueueAssignmentResult
-        QueueAssignmentResult;
-
     typedef mqbc::ClusterNodeSession::SubQueueInfo CNSSubQueueInfo;
     typedef mqbc::ClusterNodeSession::QueueState   CNSQueueState;
     typedef mqbc::ClusterNodeSession::StreamsMap   CNSStreamsMap;
@@ -483,15 +480,13 @@ class ClusterQueueHelper BSLS_KEYWORD_FINAL
                                     mqbnet::ClusterNode* primary,
                                     bmqp_ctrlmsg::PrimaryStatus::Value status);
 
-    /// Assign the queue represented by the specified `queueContext`, that
-    /// is give it an id and eventually a partition id, by initiating
-    /// assignment request communication with the leader.  Return a value
-    /// indicating whether the assignment was successful or was definitively
-    /// rejected.  This method is called regardless of proxy or member, and
-    /// leader or replica and will initiate the proper sequence of operation
-    /// based on the role of the current node within the cluster.
-    QueueAssignmentResult::Enum
-    assignQueue(const QueueContextSp& queueContext);
+    /// Try to assign the queue represented by the specified `queueContext`,
+    /// that is give it an id and eventually a partition id, by initiating
+    /// assignment request communication with the leader.   This method is
+    /// called regardless of proxy or member, and leader or replica and will
+    /// initiate the proper sequence of operation based on the role of the
+    /// current node within the cluster.
+    void assignQueue(const QueueContextSp& queueContext);
 
     /// Send a queueAssignment request to the leader, requesting assignment
     /// of the queue with the specified `uri`.  This method is called only
@@ -507,11 +502,11 @@ class ClusterQueueHelper BSLS_KEYWORD_FINAL
         const bmqt::Uri&                     uri,
         mqbnet::ClusterNode*                 responder);
 
-    /// Send a failure response for the pending contexts associated to the
-    /// states in the specified `rejected` vector.  Also remove the
-    /// associated queues from `d_queues`.
-    void processRejectedQueueAssignments(
-        const bsl::vector<QueueContext*>& rejected);
+    /// Send a failure response with the specified `status` for the pending
+    /// context associated to the states in the specified `rejected`.  Also
+    /// remove the associated queue from `d_queues`.
+    void processRejectedQueueAssignment(const QueueContext*         rejected,
+                                        const bmqp_ctrlmsg::Status& status);
 
     /// Method invoked when the queue in the specified `queueContext` has
     /// been assigned; to resume the operation on any pending contexts.

src/groups/mqb/mqbblp/mqbblp_clusterstatemanager.cpp

@@ -843,9 +843,8 @@ void ClusterStateManager::assignPartitions(
         d_clusterConfig.clusterAttributes().isCSLModeEnabled());
 }
 
-ClusterStateManager::QueueAssignmentResult::Enum
-ClusterStateManager::assignQueue(const bmqt::Uri&      uri,
-                                 bmqp_ctrlmsg::Status* status)
+bool ClusterStateManager::assignQueue(const bmqt::Uri&      uri,
+                                      bmqp_ctrlmsg::Status* status)
 {
     // executed by the cluster *DISPATCHER* thread
     // PRECONDITIONS

src/groups/mqb/mqbblp/mqbblp_clusterstatemanager.h

@@ -337,17 +337,15 @@ class ClusterStateManager BSLS_KEYWORD_FINAL
     /// Perform the actual assignment of the queue represented by the
     /// specified `uri` for a cluster member queue, that is assign it a
     /// queue key, a partition id, and some appIds; and applying the
-    /// corresponding queue assignment advisory to CSL.  Return a value
-    /// indicating whether the assignment was successful or was definitively
-    /// rejected, and populate the optionally specified `status` with a
-    /// human readable error code and string in case of failure.  This
-    /// method is called only on the leader node.
+    /// corresponding queue assignment advisory to CSL.  Return `false` in the
+    /// case of permanent failure when need to reject the assignment.  Return
+    /// `true` if the assignment is successful or can be retried.
+    /// This method is called only on the leader node.
     ///
     /// THREAD: This method is invoked in the associated cluster's
     ///         dispatcher thread.
-    QueueAssignmentResult::Enum
-    assignQueue(const bmqt::Uri&      uri,
-                bmqp_ctrlmsg::Status* status = 0) BSLS_KEYWORD_OVERRIDE;
+    bool assignQueue(const bmqt::Uri&      uri,
+                     bmqp_ctrlmsg::Status* status) BSLS_KEYWORD_OVERRIDE;
 
     /// Register a queue info for the queue with the specified `advisory`.
     /// If the specified `forceUpdate` flag is true, update queue info even if

src/groups/mqb/mqbc/mqbc_clusterstatemanager.cpp

@@ -1489,9 +1489,8 @@ void ClusterStateManager::assignPartitions(
                                   true);  // isCSLMode
 }
 
-ClusterStateManager::QueueAssignmentResult::Enum
-ClusterStateManager::assignQueue(const bmqt::Uri&      uri,
-                                 bmqp_ctrlmsg::Status* status)
+bool ClusterStateManager::assignQueue(const bmqt::Uri&      uri,
+                                      bmqp_ctrlmsg::Status* status)
 {
     // executed by the cluster *DISPATCHER* thread
     // PRECONDITIONS

src/groups/mqb/mqbc/mqbc_clusterstatemanager.h

@@ -435,15 +435,15 @@ class ClusterStateManager BSLS_KEYWORD_FINAL
     /// Perform the actual assignment of the queue represented by the
     /// specified `uri` for a cluster member queue, that is assign it a
     /// queue key, a partition id, and some appIds; and applying the
-    /// corresponding queue assignment advisory to CSL.  Return a value
-    /// indicating whether the assignment was successful or was definitively
-    /// rejected. This method is called only on the leader node.
+    /// corresponding queue assignment advisory to CSL.  Return `false` in the
+    /// case of permanent failure when need to reject the assignment.  Return
+    /// `true` if the assignment is successful or can be retried.
+    /// This method is called only on the leader node.
     ///
     /// THREAD: This method is invoked in the associated cluster's
     ///         dispatcher thread.
-    QueueAssignmentResult::Enum
-    assignQueue(const bmqt::Uri&      uri,
-                bmqp_ctrlmsg::Status* status = 0) BSLS_KEYWORD_OVERRIDE;
+    bool assignQueue(const bmqt::Uri&      uri,
+                     bmqp_ctrlmsg::Status* status) BSLS_KEYWORD_OVERRIDE;
 
     /// Register a queue info for the queue with the specified `advisory`.
     /// If the specified `forceUpdate` flag is true, update queue info even if