MB-54666: Producer doesn't enable ChangeStreams if not magma

CDC requires a magma backend.

Change-Id: If8d3f33d7e0809b4a9403d07635359200c88aed7
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/183745
Well-Formed: Restriction Checker
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: paolococchi

engines/ep/src/dcp/producer.cc

@@ -1112,6 +1112,10 @@ cb::engine_errc DcpProducer::control(uint32_t opaque,
         flatBuffersSystemEventsEnabled = true;
         return cb::engine_errc::success;
     } else if (key == DcpControlKeys::ChangeStreams && valueStr == "true") {
+        if (!engine_.getKVBucket()->getStorageProperties().canRetainHistory()) {
+            return cb::engine_errc::not_supported;
+        }
+
         changeStreams = true;
         return cb::engine_errc::success;
     }

engines/ep/tests/mock/mock_magma_kvstore.cc

@@ -13,7 +13,12 @@
 #include "kvstore/magma-kvstore/magma-memory-tracking-proxy.h"
 
 MockMagmaKVStore::MockMagmaKVStore(MagmaKVStoreConfig& config)
-    : MagmaKVStore(config) {
+    : MagmaKVStore(config),
+      storageProperties(StorageProperties::ByIdScan::Yes,
+                        StorageProperties::AutomaticDeduplication::No,
+                        StorageProperties::PrepareCounting::No,
+                        StorageProperties::CompactionStaleItemCallbacks::Yes,
+                        StorageProperties::HistoryRetentionAvailable::Yes) {
 }
 
 KVStoreIface::ReadVBStateResult MockMagmaKVStore::readVBStateFromDisk(

engines/ep/tests/mock/mock_magma_kvstore.h

@@ -65,10 +65,16 @@ class MockMagmaKVStore : public MagmaKVStore {
         fileHandleSyncStatusHook = hook;
     }
 
+    StorageProperties getStorageProperties() const override {
+        return storageProperties;
+    }
+
     TestingHook<> readVBStateFromDiskHook;
 
     std::function<int(VB::Commit&, kvstats_ctx&)> saveDocsErrorInjector;
     std::function<bool()> snapshotVBucketErrorInjector;
+
+    StorageProperties storageProperties;
 };
 
 #endif

engines/ep/tests/module_tests/dcp_single_threaded_test.cc

@@ -1132,15 +1132,19 @@ TEST_P(STDcpTest, ProducerNegotiatesFlatBuffers) {
 }
 
 TEST_P(STDcpTest, ProducerNegotiatesChangeStreams) {
-    auto* cookie = create_mock_cookie();
+    if (!isMagma()) {
+        GTEST_SKIP();
+    }
+
+    // @todo CDC: Can remove when magma has enabled history support
+    replaceMagmaKVStore();
 
+    auto* cookie = create_mock_cookie();
     const auto producer = std::make_shared<MockDcpProducer>(
             *engine, cookie, "test_producer", 0);
-
-    // Disables by default
     ASSERT_FALSE(producer->areChangeStreamsEnabled());
 
-    // DCP_CONTROL validation
+    // Value validation
     EXPECT_EQ(cb::engine_errc::invalid_arguments,
               producer->control(0, DcpControlKeys::ChangeStreams, "not-true"));
     EXPECT_FALSE(producer->areChangeStreamsEnabled());
@@ -1153,6 +1157,23 @@ TEST_P(STDcpTest, ProducerNegotiatesChangeStreams) {
     destroy_mock_cookie(cookie);
 }
 
+TEST_P(STDcpTest, ProducerNegotiatesChangeStreams_NotMagma) {
+    if (isMagma()) {
+        GTEST_SKIP();
+    }
+
+    auto* cookie = create_mock_cookie();
+    const auto producer = std::make_shared<MockDcpProducer>(
+            *engine, cookie, "test_producer", 0);
+    ASSERT_FALSE(producer->areChangeStreamsEnabled());
+
+    EXPECT_EQ(cb::engine_errc::not_supported,
+              producer->control(0, DcpControlKeys::ChangeStreams, "true"));
+    EXPECT_FALSE(producer->areChangeStreamsEnabled());
+
+    destroy_mock_cookie(cookie);
+}
+
 void STDcpTest::testConsumerNegotiatesChangeStreams(bool enabled) {
     MockDcpConnMap connMap(*engine);
     connMap.initialize();

engines/ep/tests/module_tests/dcp_stream_test.cc

@@ -5110,11 +5110,13 @@ INSTANTIATE_TEST_SUITE_P(Persistent,
                          STPassiveStreamMagmaTest,
                          STParameterizedBucketTest::magmaConfigValues(),
                          STParameterizedBucketTest::PrintToStringParamName);
-#endif /*EP_USE_MAGMA*/
 
 void CDCActiveStreamTest::SetUp() {
     STActiveStreamPersistentTest::SetUp();
 
+    // @todo CDC: Can remove as soon as magma enables history
+    replaceMagmaKVStore();
+
     CollectionsManifest manifest;
     manifest.add(CollectionEntry::historical,
                  cb::NoExpiryLimit,
@@ -5231,7 +5233,7 @@ TEST_P(CDCActiveStreamTest, CollectionNotDeduped_InMemory) {
 
 INSTANTIATE_TEST_SUITE_P(Persistent,
                          CDCActiveStreamTest,
-                         STParameterizedBucketTest::persistentConfigValues(),
+                         STParameterizedBucketTest::magmaConfigValues(),
                          STParameterizedBucketTest::PrintToStringParamName);
 
 TEST_P(CDCPassiveStreamTest, HistorySnapshotReceived) {
@@ -5277,5 +5279,7 @@ TEST_P(CDCPassiveStreamTest, HistorySnapshotReceived) {
 
 INSTANTIATE_TEST_SUITE_P(Persistent,
                          CDCPassiveStreamTest,
-                         STParameterizedBucketTest::persistentConfigValues(),
+                         STParameterizedBucketTest::magmaConfigValues(),
                          STParameterizedBucketTest::PrintToStringParamName);
+
+#endif /*EP_USE_MAGMA*/