Fix[MQB]: process assignment failure in a loop (#931)

Signed-off-by: dorjesinpo <[email protected]>

Author: dorjesinpo

src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp

@@ -168,8 +168,49 @@ void countUnconfirmed(bsls::Types::Int64* result, mqbi::Queue* queue)
         bmqp::QueueId::k_UNASSIGNED_SUBQUEUE_ID);
 }
 
+template <typename T>
+struct ConditionalAdvance {
+    bool d_doAdvance;
+
+    ConditionalAdvance()
+    : d_doAdvance(true)
+    {
+        // NOTHING
+    }
+
+    void advance(T& x)
+    {
+        if (d_doAdvance) {
+            ++x;
+        }
+        else {
+            d_doAdvance = true;
+        }
+    }
+    void release() { d_doAdvance = false; }
+};
+
 }  // close unnamed namespace
 
+// ---------------------------------------
+// struct ClusterQueueHelper::QueueContext
+// ---------------------------------------
+
+void ClusterQueueHelper::QueueContext::respond(
+    const bmqp_ctrlmsg::Status& status) const
+{
+    for (bsl::vector<OpenQueueContextSp>::const_iterator
+             cIt   = d_liveQInfo.d_pending.begin(),
+             cLast = d_liveQInfo.d_pending.end();
+         cIt != cLast;
+         ++cIt) {
+        (*cIt)->d_callback(status,
+                           0,
+                           bmqp_ctrlmsg::OpenQueueResponse(),
+                           mqbi::Cluster::OpenQueueConfirmationCookie());
+    }
+}
+
 // -------------------------------------------
 // struct ClusterQueueHelper::OpenQueueContext
 // -------------------------------------------
@@ -401,15 +442,42 @@ void ClusterQueueHelper::afterPartitionPrimaryAssignment(
     }
 }
 
-void ClusterQueueHelper::assignQueue(const QueueContextSp& queueContext)
+bool ClusterQueueHelper::assignQueueIfNeeded(
+    const QueueContextSp& queueContext_sp)
+{
+    // executed by the cluster *DISPATCHER* thread
+
+    // PRECONDITIONS
+    BSLS_ASSERT_SAFE(
+        d_cluster_p->dispatcher()->inDispatcherThread(d_cluster_p));
+
+    BSLS_ASSERT_SAFE(queueContext_sp);
+
+    if (isQueueAssigned(*queueContext_sp)) {
+        // Already assigned, nothing to do
+        return true;  // RETURN
+    }
+
+    // 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.
+
+    return assignQueue(queueContext_sp);
+}
+
+bool ClusterQueueHelper::assignQueue(const QueueContextSp& queueContext)
 {
     // executed by the cluster *DISPATCHER* thread
 
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(
         d_cluster_p->dispatcher()->inDispatcherThread(d_cluster_p));
+
+    BSLS_ASSERT_SAFE(queueContext);
     BSLS_ASSERT_SAFE(!isQueueAssigned(*queueContext));
 
+    bool result = true;
+
     if (d_cluster_p->isRemote()) {
         // Assigning a queue in a remote, is simply giving it a new queueId.
         queueContext->d_liveQInfo.d_id = getNextQueueId();
@@ -419,11 +487,12 @@ void ClusterQueueHelper::assignQueue(const QueueContextSp& queueContext)
         if (d_clusterData_p->electorInfo().isSelfLeader()) {
             bmqp_ctrlmsg::Status status(d_allocator_p);
 
-            if (!d_clusterStateManager_p->assignQueue(queueContext->uri(),
-                                                      &status)) {
-                processRejectedQueueAssignment(queueContext.get(), status);
+            result = d_clusterStateManager_p->assignQueue(queueContext->uri(),
+                                                          &status);
+
+            if (result == false) {
+                queueContext->respond(status);
             }
-            // else, all other failure are transient. will retry.
         }
         else {
             requestQueueAssignment(queueContext->uri());
@@ -439,6 +508,8 @@ void ClusterQueueHelper::assignQueue(const QueueContextSp& queueContext)
                              << queueContext->uri()
                              << "' (waiting for an ACTIVE leader).";
     }
+
+    return result;
 }
 
 void ClusterQueueHelper::requestQueueAssignment(const bmqt::Uri& uri)
@@ -618,9 +689,8 @@ void ClusterQueueHelper::onQueueAssignmentResponse(
 
                 QueueContextMapIter qit = d_queues.find(uri);
                 if (qit != d_queues.end()) {
-                    const QueueContext* rejected = qit->second.get();
-
-                    processRejectedQueueAssignment(rejected, status);
+                    qit->second->respond(status);
+                    d_queues.erase(qit);
                 }
             }
         }
@@ -3520,29 +3590,6 @@ void ClusterQueueHelper::restoreState(int partitionId)
     }
 }
 
-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));
-
-    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()
 {
     // executed by the cluster *DISPATCHER* thread
@@ -3564,9 +3611,11 @@ void ClusterQueueHelper::restoreStateRemote()
 
     // Attempt to re-issue open-queue requests for all applicable queues.
 
+    ConditionalAdvance<QueueContextMapConstIter> conditional;
+
     for (QueueContextMapConstIter cit = d_queues.cbegin();
          cit != d_queues.cend();
-         ++cit) {
+         conditional.advance(cit)) {
         const QueueContextSp& queueContext = cit->second;
         QueueLiveState&       liveQInfo    = queueContext->d_liveQInfo;
 
@@ -3581,10 +3630,13 @@ void ClusterQueueHelper::restoreStateRemote()
             continue;  // CONTINUE
         }
 
-        if (!isQueueAssigned(*queueContext.get())) {
+        if (!isQueueAssigned(*queueContext)) {
             // Queue is not assigned to a partition; get it assigned.
 
-            assignQueue(queueContext);
+            if (!assignQueue(queueContext)) {
+                conditional.release();
+                cit = d_queues.erase(cit);
+            }
 
             continue;  // CONTINUE
         }
@@ -3715,10 +3767,10 @@ void ClusterQueueHelper::restoreStateCluster(int partitionId)
         d_clusterState_p->iterateDoubleAssignments(partitionId,
                                                    doubleAssignmentVisitor);
     }
-
+    ConditionalAdvance<QueueContextMapConstIter> conditional;
     for (QueueContextMapConstIter cit = d_queues.cbegin();
          cit != d_queues.cend();
-         ++cit) {
+         conditional.advance(cit)) {
         const QueueContextSp& queueContext = cit->second;
         QueueLiveState&       liveQInfo    = queueContext->d_liveQInfo;
         if (allPartitions) {
@@ -3739,12 +3791,13 @@ void ClusterQueueHelper::restoreStateCluster(int partitionId)
                 continue;  // CONTINUE
             }
 
-            if (!isQueueAssigned(*queueContext.get())) {
-                // 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 (!isQueueAssigned(*queueContext)) {
+                // Queue is not assigned to a partition; get it assigned.
 
-                assignQueue(queueContext);
+                if (!assignQueue(queueContext)) {
+                    conditional.release();
+                    cit = d_queues.erase(cit);
+                }
 
                 continue;  // CONTINUE
             }
@@ -4738,30 +4791,35 @@ void ClusterQueueHelper::openQueue(
 
     // 'OpenQueueContext::~OpenQueueContext' decrements 'd_inFlight' counter.
 
+    bool isAssigned = false;
+
     // Check if we are already aware of the queue.
     if (queueContextIt != d_queues.end()) {
         // Already aware of the queue; but the queue may not yet have been
         // assigned.
-        context->setQueueContext(queueContextIt->second.get());
+        QueueContext& queueContext = *queueContextIt->second;
+
+        context->setQueueContext(&queueContext);
 
         // In case queue was marked for expiration, explicitly unmark it.  Note
         // that self may be a replica or a passive primary, but it's ok to
         // simply unmark the queue.  Also note that this is the necessary and
         // sufficient place to unmark a queue, as 'openQueue' is the entry
         // point.
-        queueContextIt->second->d_liveQInfo.d_queueExpirationTimestampMs = 0;
+        queueContext.d_liveQInfo.d_queueExpirationTimestampMs = 0;
 
-        if (isQueuePrimaryAvailable(*(context->queueContext()))) {
+        if (isQueuePrimaryAvailable(queueContext)) {
             // Queue is already assigned and the primary is AVAILABLE, all
             // good; move on to next step, i.e., processing the open request.
             processOpenQueueRequest(context);
+            isAssigned = true;
         }
         else {
             // The queue is already known but either not assigned, or its
             // primary is not yet available.  In both scenarios, we append that
             // context to the pending list that will be picked up and resumed
             // once the next event (primary available, queue assigned) happens.
-            QueueContext& queueContext = *(queueContextIt->second.get());
+
             queueContext.d_liveQInfo.d_pending.push_back(context);
 
             BALL_LOGTHROTTLE_INFO_BLOCK(k_MAX_INSTANT_MESSAGES,
@@ -4811,14 +4869,13 @@ void ClusterQueueHelper::openQueue(
             // advisory are received).  So to be safe, we explicitly attempt to
             // assign the queue, which is a no-op in case there is no leader.
 
-            if (!isQueueAssigned(*(queueContextIt->second))) {
-                // In CSL, unassignment is async.
-                // Since QueueUnassignmentAdvisory can contain multiple queues,
-                // canceling pending Advisory is not an option.
-                // Instead, initiate new QueueAssignemntAdvisory which must
-                // take effect after old QueueUnassignemntAdvisory.
-                assignQueue(queueContextIt->second);
-            }
+            isAssigned = isQueueAssigned(queueContext);
+
+            // In CSL, unassignment is async.
+            // Since QueueUnassignmentAdvisory can contain multiple queues,
+            // canceling pending Advisory is not an option.
+            // Instead, initiate new QueueAssignemntAdvisory which must
+            // take effect after old QueueUnassignemntAdvisory.
         }
     }
     else {
@@ -4827,14 +4884,20 @@ void ClusterQueueHelper::openQueue(
         QueueContextSp queueContext;
         queueContext.createInplace(d_allocator_p, uriKey, d_allocator_p);
 
-        d_queues[uriKey] = queueContext;
         context->setQueueContext(queueContext.get());
 
         // Register the context to the pending list.
         queueContext->d_liveQInfo.d_pending.push_back(context);
 
+        // Need to insert before calling 'assignQueue'
+        queueContextIt = d_queues.emplace(uriKey, queueContext).first;
+    }
+
+    if (!isAssigned) {
         // Initiate the assignment.
-        assignQueue(queueContext);
+        if (!assignQueue(queueContextIt->second)) {
+            d_queues.erase(queueContextIt);
+        }
     }
 }
 

src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.h

@@ -382,6 +382,10 @@ class ClusterQueueHelper BSLS_KEYWORD_FINAL
         QueueContext(const QueueContext&) BSLS_KEYWORD_DELETED;
         QueueContext& operator=(const QueueContext&) BSLS_KEYWORD_DELETED;
 
+        /// Respond to all pending OpenQueue requests with the specified
+        /// `status`.
+        void respond(const bmqp_ctrlmsg::Status& status) const;
+
         // ACCESSORS
 
         /// Return the queue uri associated with this object.
@@ -479,13 +483,14 @@ class ClusterQueueHelper BSLS_KEYWORD_FINAL
                                     mqbnet::ClusterNode* primary,
                                     bmqp_ctrlmsg::PrimaryStatus::Value status);
 
-    /// 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);
+    /// If not already assigned, try to assign the queue represented by the
+    /// specified `queueContext_sp`, that is give it an id and a partition id.
+    /// 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.  Return `false`, if the
+    /// assignment has failed, return `true` on success or on no ACTIVE leader.
+    bool assignQueueIfNeeded(const QueueContextSp& queueContext_sp);
+    bool assignQueue(const QueueContextSp& queueContext_sp);
 
     /// Send a queueAssignment request to the leader, requesting assignment
     /// of the queue with the specified `uri`.  This method is called only
@@ -501,12 +506,6 @@ class ClusterQueueHelper BSLS_KEYWORD_FINAL
         const bmqt::Uri&                     uri,
         mqbnet::ClusterNode*                 responder);
 
-    /// 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.
     void onQueueContextAssigned(const QueueContextSp& queueContext);