Fix[mqb]: to roll back, call release instead of drop (#1182)

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

Author: dorjesinpo

src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp

@@ -224,6 +224,18 @@ void ClusterQueueHelper::finishOpening(
     BSLS_ASSERT_SAFE(openQueueContext_sp);
     BSLS_ASSERT_SAFE(openQueueContext_sp->d_queueContext_p);
 
+    // First step in this routine is to update the cookie with the queue handle
+    // if 'confirmationCookie' is valid.  If this open-queue request has
+    // succeeded, this object should eventually set 'confirmationCookie' to 0
+    // (see 'onGetQueueHandleDispatched').  Note that this also applies to the
+    // 'mqba::ClientSession' case ('onQueueOpenCb').  The rough equivalent
+    // of a client session here is the cluster node session represented by
+    // 'requester'.
+
+    if (confirmationCookie) {
+        confirmationCookie->d_handle = queueHandle;
+    }
+
     const OpenQueueContext& openQueueContext = *openQueueContext_sp;
 
     openQueueContext.d_callback(status,
@@ -1408,7 +1420,7 @@ void ClusterQueueHelper::onOpenQueueResponse(
         }
 
         // 'createQueue' always calls 'onGetQueueHandle' which calls
-        // 'finishOpenQueueRequest'.
+        // 'finishOpening'.
         return;  // RETURN
     }
 
@@ -2097,14 +2109,18 @@ void ClusterQueueHelper::onOpenQueueConfirmationCookieReleased(
     // in); but upstream was a success, so we need to rollback and issue a
     // closeQueue.
 
-    BMQ_LOGTHROTTLE_WARN
+    BMQ_LOGTHROTTLE_ERROR
         << d_cluster_p->description()
         << ": OpenQueueConfirmationCookie released without "
         << "successful processing from the requester. Queue handle  "
         << "ptr [" << handle << "], client ptr [" << handle->client()
         << "], handle parameters: " << handleParameters << ".";
-    handle->clearClient(false);
-    handle->drop();
+
+    // Roll back exactly 'handleParameters'.
+    // Cannot drop the handle as it can have other substreams.
+    handle->release(handleParameters,
+                    false,
+                    mqbi::QueueHandle::HandleReleasedCallback());
 }
 
 bool ClusterQueueHelper::createQueue(
@@ -2766,18 +2782,6 @@ void ClusterQueueHelper::onGetQueueHandle(
 
     BSLS_ASSERT_SAFE(context);
 
-    // First step in this routine is to update the cookie with the queue handle
-    // if 'confirmationCookie' is valid.  If this open-queue request has
-    // succeeded, this object should eventually set 'confirmationCookie' to 0
-    // (see 'onGetQueueHandleDispatched').  Note that this also applies to the
-    // 'mqba::ClientSession' case ('onQueueOpenCb').  The rough equivalent
-    // of a client session here is the cluster node session represented by
-    // 'requester'.
-
-    if (confirmationCookie) {
-        confirmationCookie->d_handle = queueHandle;
-    }
-
     d_cluster_p->dispatcher()->execute(
         bdlf::BindUtil::bindS(d_allocator_p,
                               &ClusterQueueHelper::finishOpening,

src/groups/mqb/mqbblp/mqbblp_queue.cpp

@@ -299,12 +299,13 @@ void Queue::dropHandleDispatched(mqbi::QueueHandle* handle, bool doDeconfigure)
         isFinal = ((++citer) == handle->subStreamInfos().end());
         totalReadCount -= consumerHandleParams.readCount();
 
-        BALL_LOG_INFO << "For queue [" << handle->queue()->description()
-                      << "] and handle [" << handle->client() << ":"
-                      << handle->id() << "] " << "having [handleParamerers: "
-                      << handle->handleParameters() << "], dropping subStream "
-                      << "[" << subStreamInfo << "] having [streamParameters: "
-                      << info.d_streamParameters << "].";
+        BALL_LOG_INFO << "Dropping subStream for queue ["
+                      << handle->queue()->description() << "] and handle ["
+                      << handle->client() << ":" << handle->id()
+                      << "] handleParamerers [" << handle->handleParameters()
+                      << "], subStreamInfo [" << subStreamInfo
+                      << "], streamParameters [" << info.d_streamParameters
+                      << "].";
 
         if (doDeconfigure) {
             bmqp_ctrlmsg::StreamParameters nullStreamParameters;

src/groups/mqb/mqbblp/mqbblp_queuesessionmanager.cpp

@@ -77,6 +77,13 @@ void handleHolderDummy(BSLA_MAYBE_UNUSED const bsl::shared_ptr<void>& handle)
     // NOTHING
 }
 
+void clearClient(mqbi::QueueHandle* queueHandle)
+{
+    if (queueHandle) {
+        queueHandle->clearClient(false);
+    }
+}
+
 }  // close unnamed namespace
 
 // -------------------------
@@ -230,6 +237,8 @@ void QueueSessionManager::onQueueOpenCb(
 {
     // executed by *ANY* thread
 
+    bdlb::ScopeExitAny cleaner(bdlf::BindUtil::bind(clearClient, queueHandle));
+
     bmqu::AtomicValidatorGuard guard(validator.get());
     if (!guard.isValid()) {
         // The session was destroyed before we received the response (see
@@ -253,6 +262,8 @@ void QueueSessionManager::onQueueOpenCb(
                              responseCallback,
                              request),
         d_dispatcherClient_p);
+
+    cleaner.release();
 }
 
 void QueueSessionManager::onQueueOpenCbDispatched(
@@ -269,6 +280,8 @@ void QueueSessionManager::onQueueOpenCbDispatched(
     BSLS_ASSERT_SAFE(d_dispatcherClient_p->inDispatcherThread());
     BSLS_ASSERT_SAFE(request.choice().isOpenQueueValue());
 
+    bdlb::ScopeExitAny cleaner(bdlf::BindUtil::bind(clearClient, queueHandle));
+
     if (d_shutdownInProgress) {
         return;  // RETURN
     }
@@ -330,6 +343,8 @@ void QueueSessionManager::onQueueOpenCbDispatched(
                 cluster,
                 queueFlags);
         }
+
+        cleaner.release();
     }
 
     responseCallback(status, queueHandle, openQueueResponse);