MB-43205: Introduce DcpConsumer::allowSanitizeValueInDeletion

We will start using/testing it in follow-up patches.

Change-Id: Ied6a6c14e4d96ed5e39c81c82de8a15e7103e47a
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/143665
Tested-by: Build Bot <[email protected]>
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: paolococchi

engines/ep/src/dcp/consumer.cc

@@ -201,6 +201,8 @@ DcpConsumer::DcpConsumer(EventuallyPersistentEngine& engine,
     // Consumer needs to know if the Producer supports IncludeDeletedUserXattrs
     deletedUserXattrsNegotiation.state =
             BlockingDcpControlNegotiation::State::PendingRequest;
+
+    allowSanitizeValueInDeletion.store(config.isAllowSanitizeValueInDeletion());
 }
 
 DcpConsumer::~DcpConsumer() {

engines/ep/src/dcp/consumer.h

@@ -328,6 +328,10 @@ class DcpConsumer : public ConnHandler,
 
     void setDisconnect() override;
 
+    void setAllowSanitizeValueInDeletion(bool value) {
+        allowSanitizeValueInDeletion.store(value);
+    }
+
 protected:
     /**
      * Records when the consumer last received a message from producer.
@@ -632,6 +636,13 @@ class DcpConsumer : public ConnHandler,
      */
     size_t processBufferedMessagesBatchSize;
 
+    /**
+     * Whether this consumer should just sanitize invalid payloads in deletions
+     * or fail the operation if an invalid payload is detected.
+     * Non-const as the related configuration param is dynamic.
+     */
+    std::atomic_bool allowSanitizeValueInDeletion;
+
     static const std::string noopCtrlMsg;
     static const std::string noopIntervalCtrlMsg;
     static const std::string connBufferCtrlMsg;

engines/ep/src/dcp/dcpconnmap.cc

@@ -53,18 +53,22 @@ DcpConnMap::DcpConnMap(EventuallyPersistentEngine &e)
                     engine.getConfiguration().getDcpMinCompressionRatio());
 
     // Note: these allocations are deleted by ~Configuration
-    engine.getConfiguration().addValueChangedListener(
+    auto& config = engine.getConfiguration();
+    config.addValueChangedListener(
             "dcp_blacklist_fts_connection_logs",
             std::make_unique<DcpConfigChangeListener>(*this));
-    engine.getConfiguration().addValueChangedListener(
+    config.addValueChangedListener(
             "dcp_consumer_process_buffered_messages_yield_limit",
             std::make_unique<DcpConfigChangeListener>(*this));
-    engine.getConfiguration().addValueChangedListener(
+    config.addValueChangedListener(
             "dcp_consumer_process_buffered_messages_batch_size",
             std::make_unique<DcpConfigChangeListener>(*this));
-    engine.getConfiguration().addValueChangedListener(
+    config.addValueChangedListener(
             "dcp_idle_timeout",
             std::make_unique<DcpConfigChangeListener>(*this));
+    config.addValueChangedListener(
+            "allow_sanitize_value_in_deletion",
+            std::make_unique<DcpConfigChangeListener>(*this));
 }
 
 DcpConnMap::~DcpConnMap() {
@@ -583,6 +587,8 @@ void DcpConnMap::DcpConfigChangeListener::booleanValueChanged(const std::string&
                                                            bool value) {
     if (key == "dcp_blacklist_fts_connection_logs") {
         myConnMap.blacklistFtsConnectionLogsConfigChanged(value);
+    } else if (key == "allow_sanitize_value_in_deletion") {
+        myConnMap.consumerAllowSanitizeValueInDeletionConfigChanged(value);
     }
 }
 
@@ -636,6 +642,17 @@ void DcpConnMap::blacklistFtsConnectionLogsConfigChanged(bool newValue) {
     }
 }
 
+void DcpConnMap::consumerAllowSanitizeValueInDeletionConfigChanged(
+        bool newValue) {
+    LockHolder lh(connsLock);
+    for (const auto& cookieToConn : map_) {
+        auto* consumer = dynamic_cast<DcpConsumer*>(cookieToConn.second.get());
+        if (consumer) {
+            consumer->setAllowSanitizeValueInDeletion(newValue);
+        }
+    }
+}
+
 std::shared_ptr<ConnHandler> DcpConnMap::findByName(const std::string& name) {
     LockHolder lh(connsLock);
     for (const auto& cookieToConn : map_) {

engines/ep/src/dcp/dcpconnmap.h

@@ -199,6 +199,13 @@ class DcpConnMap : public ConnMap, public BackfillTrackingIface {
      */
     void blacklistFtsConnectionLogsConfigChanged(bool newValue);
 
+    /**
+     * Reflect the EP configuration change into all existing consumers.
+     *
+     * @param newValue
+     */
+    void consumerAllowSanitizeValueInDeletionConfigChanged(bool newValue);
+
     /**
      * @param engine The engine
      * @param cookie The cookie that identifies the connection