wait for unconfirmed before buffering confirms (#336)

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>
bloomberg · Jul 2, 2024 · c71d6c5 · c71d6c5
1 parent 82c3a71
commit c71d6c5
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 8 deletions.
diff --git a/src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp b/src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp
@@ -2633,12 +2633,20 @@ void ClusterQueueHelper::notifyQueue(QueueContext*       queueContext,
     }
 
     if (isOpen) {
-        queue->dispatcher()->execute(
-            bdlf::BindUtil::bind(&mqbi::Queue::onOpenUpstream,
-                                 queue,
-                                 generationCount,
-                                 upstreamSubQueueId),
-            queue);
+        if (generationCount == 0) {
+            BALL_LOG_INFO << d_cluster_p->description()
+                          << ": has deconfigured queue ["
+                          << queueContext->uri() << "], subStream id ["
+                          << upstreamSubQueueId << "]";
+        }
+        else {
+            queue->dispatcher()->execute(
+                bdlf::BindUtil::bind(&mqbi::Queue::onOpenUpstream,
+                                     queue,
+                                     generationCount,
+                                     upstreamSubQueueId),
+                queue);
+        }
     }
     else {
         queue->dispatcher()->execute(

diff --git a/src/integration-tests/test_puts_retransmission.py b/src/integration-tests/test_puts_retransmission.py
@@ -508,7 +508,12 @@ def test_shutdown_primary_keep_replica(self, multi_node: Cluster):
         # If shutting down primary, the replica needs to wait for new primary.
         self.active_node.wait_status(wait_leader=True, wait_ready=False)
 
-        self.inspect_results(allow_duplicates=False)
+        # Do allow duplicates for the scenario when a CONFIRM had passed Proxy
+        # but did not reach the replication.  New Primary then redelivers and
+        # the Proxy cannot detect the duplicate because it had removed the GUID
+        # upon the first CONFIRM
+
+        self.inspect_results(allow_duplicates=True)
 
     def test_shutdown_replica(self, multi_node: Cluster):
         self.setup_cluster_fanout(multi_node)
@@ -521,7 +526,12 @@ def test_shutdown_replica(self, multi_node: Cluster):
         # Because the quorum is 3, cluster is still healthy after shutting down
         # replica.
 
-        self.inspect_results(allow_duplicates=False)
+        # Do allow duplicates for the scenario when a CONFIRM had passed Proxy
+        # but did not reach the replication.  New Primary then redelivers and
+        # the Proxy cannot detect the duplicate because it had removed the GUID
+        # upon the first CONFIRM
+
+        self.inspect_results(allow_duplicates=True)
 
     def test_kill_primary_convert_replica(self, multi_node: Cluster):
         self.setup_cluster_fanout(multi_node)