From 716630043ca64b1a5cd0272744f2a1173e3dead8 Mon Sep 17 00:00:00 2001
From: Jakub Cechacek <jcechace@gmail.com>
Date: Wed, 10 Jan 2024 11:26:05 +0100
Subject: [PATCH] DBZ-7260 Added validation guard against the use of removed RS
 connection mode configuration

---
 .../connector/mongodb/MongoDbConnector.java      | 16 ++++++++++++++++
 .../connector/mongodb/MongoDbConnectorIT.java    | 13 +++++++++++++
 2 files changed, 29 insertions(+)

diff --git a/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnector.java b/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnector.java
index 29db24210..16c4d953d 100644
--- a/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnector.java
+++ b/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnector.java
@@ -37,6 +37,8 @@
 public class MongoDbConnector extends BaseSourceConnector {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(MongoDbConnector.class);
+    public static final String DEPRECATED_SHARD_CS_PARAMS_FILED = "mongodb.connection.string.shard.params";
+    public static final String DEPRECATED_CONNECTION_MODE_FILED = "mongodb.connection.mode";
 
     private Configuration config;
 
@@ -105,6 +107,20 @@ public Config validate(Map<String, String> connectorConfigs) {
     }
 
     public void validateConnection(Configuration config, ConfigValue connectionStringValidation) {
+        // Shard specific parameters shouldn't be set after RS connection mode removal
+        if (config.hasKey(DEPRECATED_SHARD_CS_PARAMS_FILED)) {
+            LOGGER.warn("Field '{}' is deprecated. Use only '{}' to set connection parameters", DEPRECATED_SHARD_CS_PARAMS_FILED,
+                    MongoDbConnectorConfig.CONNECTION_STRING.name());
+            connectionStringValidation.addErrorMessage("Deprecated field '" + DEPRECATED_SHARD_CS_PARAMS_FILED + "' is used");
+        }
+
+        // RS connection mode should not be used
+        var mode = config.getString(DEPRECATED_CONNECTION_MODE_FILED);
+        if (mode != null && mode.equals("replica_set")) {
+            LOGGER.warn("Field '{}' is deprecated. Sharded mode is now used implicitly, please remove it. ", DEPRECATED_CONNECTION_MODE_FILED);
+            connectionStringValidation.addErrorMessage("Deprecated field '" + DEPRECATED_CONNECTION_MODE_FILED + "' is used set to removed 'replica_set' value");
+        }
+
         MongoDbConnectionContext connectionContext = new MongoDbConnectionContext(config);
 
         try {
diff --git a/debezium-connector-mongodb/src/test/java/io/debezium/connector/mongodb/MongoDbConnectorIT.java b/debezium-connector-mongodb/src/test/java/io/debezium/connector/mongodb/MongoDbConnectorIT.java
index 4fea58b60..39b9a0147 100644
--- a/debezium-connector-mongodb/src/test/java/io/debezium/connector/mongodb/MongoDbConnectorIT.java
+++ b/debezium-connector-mongodb/src/test/java/io/debezium/connector/mongodb/MongoDbConnectorIT.java
@@ -114,6 +114,19 @@ public void shouldFailToValidateInvalidConfiguration() {
         assertNoConfigurationErrors(result, CommonConnectorConfig.TOMBSTONES_ON_DELETE);
     }
 
+    @Test
+    public void shouldFailToValidateWithReplicaSetModeParams() {
+        config = TestHelper.getConfiguration(mongo)
+                .edit()
+                .with(MongoDbConnector.DEPRECATED_SHARD_CS_PARAMS_FILED, "readPreference=primary")
+                .with(MongoDbConnector.DEPRECATED_CONNECTION_MODE_FILED, "replica_set")
+                .build();
+        MongoDbConnector connector = new MongoDbConnector();
+        Config result = connector.validate(config.asMap());
+
+        assertConfigurationErrors(result, MongoDbConnectorConfig.CONNECTION_STRING, 2);
+    }
+
     @Test
     public void shouldThrowExceptionWhenFieldExcludeListDatabasePartIsOnlyProvided() {
         shouldValidateFilterFieldConfiguration(MongoDbConnectorConfig.FIELD_EXCLUDE_LIST, "inventory", 1);