MB-71134: Close DCP consumer stream after test

Suspect intermitent failure is due to a race with previous
test state. Noted in logs when test fails a WARNING

"Failing to add passive stream, as one already exists for the vbucket"

Then a disconnect is sent and test fails.

Change-Id: I73102d9410d6838f57a8faff96818ce3c554caed
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/244899
Reviewed-by: Paolo Cocchi <paolo.cocchi@couchbase.com>
Reviewed-by: Trond Norbye <trond.norbye@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: jimwwalker

protocol/connection/client_connection.cc

@@ -2167,6 +2167,12 @@ void MemcachedConnection::dcpAddStream(Vbid vbid,
     sendCommand(BinprotDcpAddStreamCommand{flags, vbid});
 }
 
+void MemcachedConnection::dcpCloseStream(Vbid vbid) {
+    BinprotGenericCommand cmd{cb::mcbp::ClientOpcode::DcpCloseStream};
+    cmd.setVBucket(vbid);
+    sendCommand(cmd);
+}
+
 void MemcachedConnection::dcpStreamRequestResponse(
         uint32_t opaque,
         const std::vector<std::pair<uint64_t, uint64_t>>& failovers) {

protocol/connection/client_connection.h

@@ -1126,6 +1126,9 @@ class MemcachedConnection {
                           uint64_t snapEnd,
                           const nlohmann::json& value);
 
+    /// Send a DcpCloseStream command for the given vbucket on this connection.
+    void dcpCloseStream(Vbid vbid);
+
     /* following dcp functions are for working with a consumer */
     void dcpAddStream(Vbid vbid, cb::mcbp::DcpAddStreamFlag flags = {});
 

tests/testapp/testapp_dcp_consumer.cc

@@ -86,6 +86,11 @@ class DcpConsumerAckTest
         fullEviction = evictionPolicy == "full_eviction";
     }
 
+    void TearDown() override {
+        conn->dcpCloseStream(Vbid(0));
+        TestappTest::TearDown();
+    }
+
     void setupConsumer(
             std::string_view name,
             const std::vector<std::pair<std::string, std::string>>& controls) {