[Improve] [Connector-V2] Optimize milvus-connector config code (#7658)

seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/config/MilvusCommonConfig.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.milvus.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+public abstract class MilvusCommonConfig {
+
+    public static final String CONNECTOR_IDENTITY = "Milvus";
+
+    public static final Option<String> URL =
+            Options.key("url")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Milvus public endpoint");
+
+    public static final Option<String> TOKEN =
+            Options.key("token")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Milvus token for authentication");
+}

seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/config/MilvusSinkConfig.java

@@ -30,21 +30,7 @@
 import static org.apache.seatunnel.api.sink.DataSaveMode.DROP_DATA;
 import static org.apache.seatunnel.api.sink.DataSaveMode.ERROR_WHEN_DATA_EXISTS;
 
-public class MilvusSinkConfig {
-
-    public static final String CONNECTOR_IDENTITY = "Milvus";
-
-    public static final Option<String> URL =
-            Options.key("url")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("Milvus public endpoint");
-
-    public static final Option<String> TOKEN =
-            Options.key("token")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("Milvus token for authentication");
+public class MilvusSinkConfig extends MilvusCommonConfig {
 
     public static final Option<String> DATABASE =
             Options.key("database")

seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/config/MilvusSourceConfig.java

@@ -20,46 +20,17 @@
 import org.apache.seatunnel.api.configuration.Option;
 import org.apache.seatunnel.api.configuration.Options;
 
-import java.util.Map;
-
-public class MilvusSourceConfig {
-
-    public static final Option<String> URL =
-            Options.key("url")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("Milvus public endpoint");
-
-    public static final Option<String> TOKEN =
-            Options.key("token")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("Milvus token for authentication");
+public class MilvusSourceConfig extends MilvusCommonConfig {
 
     public static final Option<String> DATABASE =
             Options.key("database")
                     .stringType()
-                    .noDefaultValue()
+                    .defaultValue("default")
                     .withDescription("database");
 
     public static final Option<String> COLLECTION =
             Options.key("collection")
                     .stringType()
                     .noDefaultValue()
                     .withDescription("Milvus collection to read");
-    public static final Option<Map<String, String>> COLLECTION_RENAME =
-            Options.key("collection_rename")
-                    .mapType()
-                    .noDefaultValue()
-                    .withDescription("rename collections");
-    public static final Option<Integer> BATCH_SIZE =
-            Options.key("batch_size")
-                    .intType()
-                    .defaultValue(1000)
-                    .withDescription("writer batch size");
-    public static final Option<Integer> RATE_LIMIT =
-            Options.key("rate_limit")
-                    .intType()
-                    .defaultValue(1000000)
-                    .withDescription("writer rate limit");
 }

seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/source/MilvusSource.java

@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package org.apache.seatunnel.connectors.seatunnel.milvus.source;
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
@@ -10,7 +27,8 @@
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
-import org.apache.seatunnel.connectors.seatunnel.milvus.utils.MilvusConvertUtils;
+import org.apache.seatunnel.connectors.seatunnel.milvus.config.MilvusSourceConfig;
+import org.apache.seatunnel.connectors.seatunnel.milvus.convert.MilvusConvertUtils;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -24,10 +42,9 @@ public class MilvusSource
     private final ReadonlyConfig config;
     private final Map<TablePath, CatalogTable> sourceTables;
 
-    public MilvusSource(ReadonlyConfig sourceConfing) {
-        this.config = sourceConfing;
-        MilvusConvertUtils milvusConvertUtils = new MilvusConvertUtils(sourceConfing);
-        this.sourceTables = milvusConvertUtils.getSourceTables();
+    public MilvusSource(ReadonlyConfig sourceConfig) {
+        this.config = sourceConfig;
+        this.sourceTables = MilvusConvertUtils.getSourceTables(config);
     }
 
     @Override
@@ -61,6 +78,6 @@ public SourceSplitEnumerator<MilvusSourceSplit, MilvusSourceState> restoreEnumer
 
     @Override
     public String getPluginName() {
-        return "Milvus";
+        return MilvusSourceConfig.CONNECTOR_IDENTITY;
     }
 }