Refactor[MQB]: simplify queue creation/registration (#827)

dorjesinpo · web-flow · commit c4c3bd89dca0 · 2025-07-11T13:02:58.000-04:00
Signed-off-by: dorjesinpo &lt;129227380+dorjesinpo@users.noreply.github.com&gt;
diff --git a/src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp b/src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp
@@ -2039,19 +2039,12 @@ bsl::shared_ptr<mqbi::Queue> ClusterQueueHelper::createQueueFactory(
                                    d_allocator_p),
         d_allocator_p);
 
+    // Create Local/Remote queue flavor
     if (!isPrimary) {
         queueSp->createRemote(
             openQueueResponse.deduplicationTimeMs(),
             d_clusterData_p->clusterConfig().queueOperations().ackWindowSize(),
             &d_clusterData_p->stateSpPool());
-
-        if (context.d_domain_p->registerQueue(errorDescription, queueSp) !=
-            0) {
-            return 0;  // RETURN
-        }
-
-        queueContext->d_liveQInfo.d_queue_sp         = queueSp;
-        d_queuesById[queueContext->d_liveQInfo.d_id] = queueContext;
     }
     else {
         // This is the primary of the queue.
@@ -2084,24 +2077,62 @@ bsl::shared_ptr<mqbi::Queue> ClusterQueueHelper::createQueueFactory(
         // Queue must have been registered with storage manager before
         // registering it with the domain, otherwise Queue.configure() will
         // fail.
+    }
 
-        if (context.d_domain_p->registerQueue(errorDescription, queueSp) !=
-            0) {
-            return 0;  // RETURN
-        }
+    // Register this queue to the dispatcher.
+    if (d_cluster_p->isRemote()) {
+        d_cluster_p->dispatcher()->registerClient(
+            queueSp.get(),
+            mqbi::DispatcherClientType::e_QUEUE);
+    }
+    else {
+        d_cluster_p->dispatcher()->registerClient(
+            queueSp.get(),
+            mqbi::DispatcherClientType::e_QUEUE,
+            d_storageManager_p->processorForPartition(
+                queueContext->partitionId()));
+    }
+
+    // Configure the queue
+    bdlma::LocalSequentialAllocator<1024> localAllocator(d_allocator_p);
+    bmqu::MemOutStream                    error(&localAllocator);
 
-        queueContext->d_liveQInfo.d_queue_sp = queueSp;
-        // No need to insert in d_queuesById since those queues will never
-        // be looked up by id (and all have k_PRIMARY_QUEUE_ID id).
+    int rc = queueSp->configure(error,
+                                false,  // isReconfigure
+                                true);  // wait
+
+    if (rc != 0) {
+        // Queue.configure() failed.
+
+        BALL_LOG_ERROR << "Failure configuring queue '" << queueContext->uri()
+                       << "': " << error.str() << ".";
+
+        errorDescription << error.str();
+
+        // Discard the queue.
+        return 0;  // RETURN
+    }
+
+    if (context.d_domain_p->registerQueue(queueSp) != 0) {
+        // Discard the queue.
+        return 0;  // RETURN
+    }
+
+    queueContext->d_liveQInfo.d_queue_sp = queueSp;
+
+    if (!isPrimary) {
+        d_queuesById[queueContext->d_liveQInfo.d_id] = queueContext;
     }
+    // else, no need to insert in d_queuesById since those queues will never
+    // be looked up by id (and all have k_PRIMARY_QUEUE_ID id).
 
     if (!d_cluster_p->isRemote()) {
         d_clusterState_p->updatePartitionNumActiveQueues(
             queueContext->partitionId(),
             1);
     }
 
-    return bsl::shared_ptr<mqbi::Queue>(queueSp);
+    return queueSp;
 }
 
 void ClusterQueueHelper::onHandleReleased(
diff --git a/src/groups/mqb/mqbblp/mqbblp_domain.cpp b/src/groups/mqb/mqbblp/mqbblp_domain.cpp
@@ -475,8 +475,7 @@ void Domain::openQueue(
                              callback));
 }
 
-int Domain::registerQueue(bsl::ostream&                       errorDescription,
-                          const bsl::shared_ptr<mqbi::Queue>& queueSp)
+int Domain::registerQueue(const bsl::shared_ptr<mqbi::Queue>& queueSp)
 {
     // executed by the associated CLUSTER's DISPATCHER thread
 
@@ -486,9 +485,8 @@ int Domain::registerQueue(bsl::ostream&                       errorDescription,
 
     enum RcEnum {
         // Value for the various RC error categories
-        rc_SUCCESS              = 0,
-        rc_ALREADY_REGISTERED   = -1,
-        rc_CONFIGURATION_FAILED = -2
+        rc_SUCCESS            = 0,
+        rc_ALREADY_REGISTERED = -1
     };
 
     // As part of registering the 'queue' with the domain, 'queue' will also be
@@ -519,30 +517,6 @@ int Domain::registerQueue(bsl::ostream&                       errorDescription,
 
         d_queues[queueSp->uri().queue()] = queueSp;
     }
-    bdlma::LocalSequentialAllocator<1024> localAllocator(d_allocator_p);
-    bmqu::MemOutStream                    error(&localAllocator);
-
-    int rc = queueSp->configure(error,
-                                false,  // isReconfigure
-                                true);  // wait
-    if (rc != 0) {
-        // Queue.configure() failed, need to rollback.
-
-        BALL_LOG_ERROR << "Failure configuring queue in the domain '" << d_name
-                       << "' "
-                       << "[canonicalURI: " << queueSp->uri().canonical()
-                       << ", qId: " << bmqp::QueueId::QueueIdInt(queueSp->id())
-                       << "]: " << error.str() << ".";
-
-        errorDescription << error.str();
-
-        bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCK
-        QueueMap::const_iterator it = d_queues.find(queueSp->uri().queue());
-        BSLS_ASSERT_SAFE(it != d_queues.end());
-        d_queues.erase(it);
-
-        return rc * 10 + rc_CONFIGURATION_FAILED;  // RETURN
-    }
 
     BALL_LOG_INFO << "Registered queue to domain '" << d_name << "' "
                   << "[canonicalURI: " << queueSp->uri().canonical()
diff --git a/src/groups/mqb/mqbblp/mqbblp_domain.h b/src/groups/mqb/mqbblp/mqbblp_domain.h
@@ -267,11 +267,9 @@ class Domain BSLS_KEYWORD_FINAL : public mqbi::Domain {
                    const mqbi::Domain::OpenQueueCallback&     callback)
         BSLS_KEYWORD_OVERRIDE;
 
-    /// Take ownership of and configure the specified `queue`.  Return 0 on
-    /// success, or a non-zero return code otherwise, populating the
-    /// specified `errorDescription` with a description of the failure.
-    int registerQueue(bsl::ostream&                       errorDescription,
-                      const bsl::shared_ptr<mqbi::Queue>& queueSp)
+    /// Take ownership of the specified `queue`.  Return 0 on success, or a
+    /// non-zero return code otherwise.
+    int registerQueue(const bsl::shared_ptr<mqbi::Queue>& queueSp)
         BSLS_KEYWORD_OVERRIDE;
 
     /// Reverse method of `registerQueue`, invoked when the last
diff --git a/src/groups/mqb/mqbblp/mqbblp_queue.cpp b/src/groups/mqb/mqbblp/mqbblp_queue.cpp
@@ -448,10 +448,6 @@ Queue::Queue(const bmqt::Uri&                          uri,
         .setMiscWorkThreadPool(threadPool)
         .setRoutingConfig(routingCfg)
         .setMessageThrottleConfig(messageThrottleConfig);
-
-    // Have to set up dispatcher in constructor so we can check if we are
-    // in the correct thread in `mqbblp::Queue::configure()`.
-    dispatcherClientData().setDispatcher(domain->cluster()->dispatcher());
 }
 
 Queue::~Queue()
@@ -584,26 +580,7 @@ int Queue::configure(bsl::ostream& errorDescription,
                      bool          isReconfigure,
                      bool          wait)
 {
-    // executed by the cluster *DISPATCHER* thread
-
-    // PRECONDITIONS
-    BSLS_ASSERT_SAFE(
-        dispatcher()->inDispatcherThread(d_state.domain()->cluster()));
-
-    if (!isReconfigure) {
-        // Register this queue to the dispatcher.
-        if (d_state.domain()->cluster()->isRemote()) {
-            dispatcher()->registerClient(this,
-                                         mqbi::DispatcherClientType::e_QUEUE);
-        }
-        else {
-            dispatcher()->registerClient(
-                this,
-                mqbi::DispatcherClientType::e_QUEUE,
-                d_state.storageManager()->processorForPartition(
-                    d_state.partitionId()));
-        }
-    }
+    // executed by *ANY* thread
 
     // Enqueue a configure callback in the queue-dispatcher thread.
     int result = 0;
diff --git a/src/groups/mqb/mqbblp/mqbblp_queueconsumptionmonitor.t.cpp b/src/groups/mqb/mqbblp/mqbblp_queueconsumptionmonitor.t.cpp
@@ -246,7 +246,7 @@ Test::Test()
     bslma::ManagedPtr<mqbi::Queue> queueMp(&d_queue,
                                            0,
                                            bslma::ManagedPtrUtil::noOpDeleter);
-    d_domain.registerQueue(errorDescription, queueMp);
+    d_domain.registerQueue(queueMp);
 
     mqbconfm::Storage config;
     mqbconfm::Limits  limits;
diff --git a/src/groups/mqb/mqbblp/mqbblp_queueenginetester.cpp b/src/groups/mqb/mqbblp/mqbblp_queueenginetester.cpp
@@ -470,7 +470,13 @@ void QueueEngineTester::init(const mqbconfm::Domain& domainConfig,
 
     // Register queue in domain
     bslma::ManagedPtr<mqbi::Queue> queueMp(d_mockQueue_sp.managedPtr());
-    rc = d_mockDomain_mp->registerQueue(errorDescription, queueMp);
+
+    rc = queueMp->configure(errorDescription,
+                            false,  // isReconfigure
+                            true);  // wait
+    BSLS_ASSERT_OPT(rc == 0);
+
+    rc = d_mockDomain_mp->registerQueue(queueMp);
     BSLS_ASSERT_OPT(rc == 0);
 
     // VALIDATION
diff --git a/src/groups/mqb/mqbi/mqbi_domain.h b/src/groups/mqb/mqbi/mqbi_domain.h
@@ -163,12 +163,9 @@ class Domain {
         const bmqp_ctrlmsg::QueueHandleParameters&          handleParameters,
         const OpenQueueCallback&                            callback) = 0;
 
-    /// Take ownership of the specified `queue`, and eventually configure
-    /// it.  Return 0 on success, or a non-zero return code otherwise,
-    /// populating the specified `errorDescription` with a description of
-    /// the failure.
-    virtual int registerQueue(bsl::ostream&                 errorDescription,
-                              const bsl::shared_ptr<Queue>& queueSp) = 0;
+    /// Take ownership of the specified `queue`. Return 0 on success, or a
+    /// non-zero return code otherwise.
+    virtual int registerQueue(const bsl::shared_ptr<Queue>& queueSp) = 0;
 
     /// Reverse method of `registerQueue`, invoked when the last
     /// `QueueHandle` associated to the specified `queue` has been deleted
diff --git a/src/groups/mqb/mqbmock/mqbmock_domain.cpp b/src/groups/mqb/mqbmock/mqbmock_domain.cpp
@@ -99,21 +99,13 @@ void Domain::openQueue(
     // NOTHING
 }
 
-int Domain::registerQueue(bsl::ostream&                       errorDescription,
-                          const bsl::shared_ptr<mqbi::Queue>& queueSp)
+int Domain::registerQueue(const bsl::shared_ptr<mqbi::Queue>& queueSp)
 {
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(queueSp && "'queue' must not be null");
     BSLS_ASSERT_SAFE(lookupQueue(0, queueSp->uri()) != 0 &&
                      "'queue' already registered with the domain");
 
-    int rc = queueSp->configure(errorDescription,
-                                false,  // isReconfigure
-                                true);  // wait
-    if (rc != 0) {
-        return rc;  // RETURN
-    }
-
     d_queues[queueSp->uri().queue()] = queueSp;
 
     return 0;
diff --git a/src/groups/mqb/mqbmock/mqbmock_domain.h b/src/groups/mqb/mqbmock/mqbmock_domain.h
@@ -182,12 +182,9 @@ class Domain : public mqbi::Domain {
                    const mqbi::Domain::OpenQueueCallback&     callback)
         BSLS_KEYWORD_OVERRIDE;
 
-    /// Take ownership of the specified `queue`, and eventually configure
-    /// it.  Return 0 on success, or a non-zero return code otherwise,
-    /// populating the specified `errorDescription` with a description of
-    /// the failure.
-    int registerQueue(bsl::ostream&                       errorDescription,
-                      const bsl::shared_ptr<mqbi::Queue>& queueSp)
+    /// Take ownership of the specified `queue`.  Return 0 on success, or a
+    /// non-zero return code otherwise.
+    int registerQueue(const bsl::shared_ptr<mqbi::Queue>& queueSp)
         BSLS_KEYWORD_OVERRIDE;
 
     /// Reverse method of `registerQueue`, invoked when the last
