Fix the review problem

apache · Jun 1, 2023 · 1adb461 · 1adb461
1 parent 60fe820
commit 1adb461
Show file tree

Hide file tree

Showing 5 changed files with 21 additions and 10 deletions.
diff --git a/docs/en/connector-v2/sink/MongoDB.md b/docs/en/connector-v2/sink/MongoDB.md
@@ -83,6 +83,7 @@ Sink Options
 
 > 1.The data flushing logic of the MongoDB Sink Connector is jointly controlled by three parameters: `buffer-flush.max-rows`, `buffer-flush.interval`, and `checkpoint.interval`.
 > Data flushing will be triggered if any of these conditions are met.<br/>
+> 2.Compatible with the historical parameter `upsert-key`. If `upsert-key` is set, please do not set `primary-key`.<br/>
 
 How to Create a MongoDB Data Synchronization Jobs
 -------------------------------------------------

diff --git a/...src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/config/MongodbConfig.java b/...src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/config/MongodbConfig.java
@@ -147,5 +147,6 @@ public class MongodbConfig {
                     .listType()
                     .noDefaultValue()
                     .withDescription(
-                            "The primary keys for upsert/update. Keys are in csv format for properties.");
+                            "The primary keys for upsert/update. Keys are in csv format for properties.")
+                    .withFallbackKeys("upsert-key");
 }
diff --git a/...c/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongoKeyExtractor.java b/...c/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongoKeyExtractor.java
@@ -31,7 +31,7 @@ public class MongoKeyExtractor implements SerializableFunction<BsonDocument, Bso
     private final String[] upsertKey;
 
     public MongoKeyExtractor(MongodbWriterOptions options) {
-        upsertKey = options.getUpsertKey();
+        upsertKey = options.getPrimaryKey();
     }
 
     @Override

diff --git a/...odb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSink.java b/...odb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSink.java
@@ -34,6 +34,7 @@
 import com.google.auto.service.AutoService;
 
 import java.io.IOException;
+import java.util.List;
 
 import static org.apache.seatunnel.connectors.seatunnel.mongodb.config.MongodbConfig.CONNECTOR_IDENTITY;
 
@@ -66,11 +67,19 @@ public void prepare(Config pluginConfig) throws PrepareFailException {
                         pluginConfig.getLong(MongodbConfig.BUFFER_FLUSH_INTERVAL.key()));
             }
             if (pluginConfig.hasPath(MongodbConfig.PRIMARY_KEY.key())) {
-                builder.withUpsertKey(
+                builder.withPrimaryKey(
                         pluginConfig
                                 .getStringList(MongodbConfig.PRIMARY_KEY.key())
                                 .toArray(new String[0]));
             }
+            List<String> fallbackKeys = MongodbConfig.PRIMARY_KEY.getFallbackKeys();
+            fallbackKeys.forEach(
+                    key -> {
+                        if (pluginConfig.hasPath(key)) {
+                            builder.withPrimaryKey(
+                                    pluginConfig.getStringList(key).toArray(new String[0]));
+                        }
+                    });
             if (pluginConfig.hasPath(MongodbConfig.UPSERT_ENABLE.key())) {
                 builder.withUpsertEnable(
                         pluginConfig.getBoolean(MongodbConfig.UPSERT_ENABLE.key()));

diff --git a/...ain/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriterOptions.java b/...ain/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriterOptions.java
@@ -38,7 +38,7 @@ public class MongodbWriterOptions implements Serializable {
 
     protected final boolean upsertEnable;
 
-    protected final String[] upsertKey;
+    protected final String[] primaryKey;
 
     protected final int retryMax;
 
@@ -51,7 +51,7 @@ public MongodbWriterOptions(
             int flushSize,
             Long batchIntervalMs,
             boolean upsertEnable,
-            String[] upsertKey,
+            String[] primaryKey,
             int retryMax,
             Long retryInterval) {
         this.connectString = connectString;
@@ -60,7 +60,7 @@ public MongodbWriterOptions(
         this.flushSize = flushSize;
         this.batchIntervalMs = batchIntervalMs;
         this.upsertEnable = upsertEnable;
-        this.upsertKey = upsertKey;
+        this.primaryKey = primaryKey;
         this.retryMax = retryMax;
         this.retryInterval = retryInterval;
     }
@@ -83,7 +83,7 @@ public static class Builder {
 
         protected boolean upsertEnable;
 
-        protected String[] upsertKey;
+        protected String[] primaryKey;
 
         protected int retryMax;
 
@@ -119,8 +119,8 @@ public Builder withUpsertEnable(boolean upsertEnable) {
             return this;
         }
 
-        public Builder withUpsertKey(String[] upsertKey) {
-            this.upsertKey = upsertKey;
+        public Builder withPrimaryKey(String[] upsertKey) {
+            this.primaryKey = primaryKey;
             return this;
         }
 
@@ -142,7 +142,7 @@ public MongodbWriterOptions build() {
                     flushSize,
                     batchIntervalMs,
                     upsertEnable,
-                    upsertKey,
+                    primaryKey,
                     retryMax,
                     retryInterval);
         }