MB-35427: Remove ack for correct consumer

When we set the ActiveStream to dead we attempt to remove any
queuedSeqnoAck from the DM's map. We currently attempt to remove
the ack with the name of the ActiveStream which is not correct. We
should instead remove the ack with the name of the consumer which is
stored in the DcpProducer.

Change-Id: Ie4308b114542d61f2ffd91f6e65375f88114fe54
Reviewed-on: http://review.couchbase.org/112919
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Dave Rigby <[email protected]>

Author: BenHuddleston

engines/ep/src/dcp/active_stream.cc

@@ -1185,11 +1185,33 @@ uint32_t ActiveStream::setDead(end_stream_status_t status) {
         if (!vb) {
             return 0;
         }
+
+        // Get the consumer name from the producer so that we can clear the
+        // correct ack
+        std::string consumerName;
+        {
+            auto p = producerPtr.lock();
+            if (!p) {
+                log(spdlog::level::warn,
+                    "Producer could not be locked when"
+                    "attempting to clear queued seqno acks");
+                return 0;
+            }
+            consumerName = p->getConsumerName();
+        }
+
+        if (consumerName.empty()) {
+            log(spdlog::level::warn,
+                "Consumer name not found for producer when"
+                "attempting to clear queued seqno acks");
+            return 0;
+        }
+
         // Take the vb state lock so that we don't change the state of
         // this vb
         folly::SharedMutex::ReadHolder vbStateLh(vb->getStateLock());
         if (vb->getState() == vbucket_state_active) {
-            vb->removeQueuedAckFromDM(name_);
+            vb->removeQueuedAckFromDM(consumerName);
         }
     }
 

engines/ep/src/dcp/producer.cc

@@ -1827,3 +1827,7 @@ end_stream_status_t DcpProducer::mapEndStreamStatus(
     }
     return status;
 }
+
+std::string DcpProducer::getConsumerName() const {
+    return consumerName;
+}

engines/ep/src/dcp/producer.h

@@ -329,6 +329,8 @@ class DcpProducer : public ConnHandler,
     std::shared_ptr<StreamContainer<std::shared_ptr<Stream>>> findStreams(
             Vbid vbid);
 
+    std::string getConsumerName() const;
+
 protected:
     /** We may disconnect if noop messages are enabled and the last time we
      *  received any message (including a noop) exceeds the dcpTimeout.

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -48,7 +48,8 @@ void DurabilityActiveStreamTest::setUp(bool startCheckpointProcessorTask) {
 
     // Enable SyncReplication and flow-control (Producer BufferLog)
     setupProducer({{"enable_synchronous_replication", "true"},
-                   {"connection_buffer_size", "52428800"}},
+                   {"connection_buffer_size", "52428800"},
+                   {"consumer_name", "test_consumer"}},
                   startCheckpointProcessorTask);
     ASSERT_TRUE(stream->public_supportSyncReplication());
 }
@@ -474,7 +475,7 @@ TEST_P(DurabilityActiveStreamTest, RemoveUnknownSeqnoAckAtDestruction) {
     // Our topology gives replica name as "replica" an our producer/stream has
     // name "test_producer". Simulate a seqno ack by calling the vBucket level
     // function.
-    stream->seqnoAck("test_producer", 1);
+    stream->seqnoAck(producer->getConsumerName(), 1);
 
     // An unknown seqno ack should not have committed the item
     EXPECT_EQ(0, vb->getNumItems());
@@ -487,7 +488,7 @@ TEST_P(DurabilityActiveStreamTest, RemoveUnknownSeqnoAckAtDestruction) {
     // connections. We verify that the seqno ack does not exist in the map
     // by performing the topology change that would commit the prepare if it
     // did.
-    EXPECT_EQ(ENGINE_SUCCESS, stream->seqnoAck("test_producer", 1));
+    EXPECT_EQ(ENGINE_SUCCESS, stream->seqnoAck(producer->getConsumerName(), 1));
 
     // If the seqno ack still existed in the queuedSeqnoAcks map then it would
     // result in a commit on topology change
@@ -496,11 +497,60 @@ TEST_P(DurabilityActiveStreamTest, RemoveUnknownSeqnoAckAtDestruction) {
             vbucket_state_active,
             {{"topology",
               nlohmann::json::array(
-                      {{"active", "replica1", "test_producer"}})}});
+                      {{"active", "replica1", producer->getConsumerName()}})}});
 
     EXPECT_EQ(0, vb->getNumItems());
 }
 
+TEST_P(DurabilityActiveStreamTest, RemoveCorrectQueuedAckAtStreamSetDead) {
+    auto vb = engine->getVBucket(vbid);
+
+    const auto key = makeStoredDocKey("key");
+    const auto& value = "value";
+    auto item = makePendingItem(
+            key,
+            value,
+            cb::durability::Requirements(cb::durability::Level::Majority,
+                                         1 /*timeout*/));
+    VBQueueItemCtx ctx;
+    ctx.durability =
+            DurabilityItemCtx{item->getDurabilityReqs(), nullptr /*cookie*/};
+
+    EXPECT_EQ(MutationStatus::WasClean, public_processSet(*vb, *item, ctx));
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    // We don't include prepares in the numItems stat (should not exist in here)
+    EXPECT_EQ(0, vb->getNumItems());
+
+    // Our topology gives replica name as "replica" an our producer/stream has
+    // name "test_producer". Simulate a seqno ack by calling the vBucket level
+    // function.
+    stream->seqnoAck(producer->getConsumerName(), 1);
+
+    // Disconnect the ActiveStream. Should remove the queued seqno ack
+    stream->setDead(END_STREAM_DISCONNECTED);
+
+    stream = std::make_shared<MockActiveStream>(engine.get(),
+                                                producer,
+                                                0 /*flags*/,
+                                                0 /*opaque*/,
+                                                *vb,
+                                                0 /*st_seqno*/,
+                                                ~0 /*en_seqno*/,
+                                                0x0 /*vb_uuid*/,
+                                                0 /*snap_start_seqno*/,
+                                                ~0 /*snap_end_seqno*/);
+    producer->createCheckpointProcessorTask();
+    producer->scheduleCheckpointProcessorTask();
+    stream->setActive();
+
+    // Should not throw a monotonic exception as the ack should have been
+    // removed by setDead.
+    stream->seqnoAck(producer->getConsumerName(), 1);
+
+    producer->cancelCheckpointCreatorTask();
+}
+
 void DurabilityActiveStreamTest::setUpSendSetInsteadOfCommitTest() {
     auto vb = engine->getVBucket(vbid);
 

engines/ep/tests/module_tests/dcp_stream_test.cc

@@ -1427,8 +1427,11 @@ void SingleThreadedActiveStreamTest::setupProducer(
     // We don't set the startTask flag here because we will create the task
     // manually. We do this because the producer actually creates the task on
     // StreamRequest which we do not do because we want a MockActiveStream.
-    producer = std::make_shared<MockDcpProducer>(
-            *engine, cookie, "test_producer", flags, false /*startTask*/);
+    producer = std::make_shared<MockDcpProducer>(*engine,
+                                                 cookie,
+                                                 "test_producer->test_consumer",
+                                                 flags,
+                                                 false /*startTask*/);
 
     if (startCheckpointProcessorTask &&
         !producer->getCheckpointSnapshotTask()) {