[Feature][Connector-v2][Mongodb]Refactor mongodb connector #4620

MonsterChenzhuo · 2023-04-19T08:32:32Z

Purpose of this pull request

Check list

Code changed are covered with tests, or it does not need tests for reason:
If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update change log that in connector document. For more details you can refer to connector-v2
2. Update plugin-mapping.properties and add new connector information in it
3. Update the pom file of seatunnel-dist
Update the release-note.

MonsterChenzhuo · 2023-04-19T08:37:41Z

I will commit the changes to the release-note.md file after the PR review is completed, in order to prevent conflicts that may cause the CI to not run.

MonsterChenzhuo · 2023-04-19T08:46:49Z

cc @TyrantLucifer @hailin0 PTAL, thanks.

TyrantLucifer · 2023-04-19T09:46:55Z

...ector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java

@@ -106,75 +83,6 @@ public void initConnection() {
        client = MongoClients.create(url);
    }



Add more test case to verify refactor is make sense.

TyrantLucifer · 2023-04-19T09:47:00Z

...ector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java

 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;

 import static java.net.HttpURLConnection.HTTP_OK;
 import static java.net.HttpURLConnection.HTTP_UNAUTHORIZED;

 @Slf4j
+@DisabledOnContainer(
+        value = {},
+        type = {EngineType.FLINK, EngineType.SPARK})


Why disable spark and flink?

@TyrantLucifer I apologize for the inconvenience. Due to the fact that the Flink 1.13 image cannot run on the Mac M1, I have encountered issues with running Flink CI locally, even though I have successfully validated it with Zeta, Flink, and Spark. When I submit the code directly to GitHub, it often fails for various reasons. My plan is to first use this annotation to limit the scope and get Flink running smoothly. After that, I will remove the annotation.

TyrantLucifer · 2023-04-20T03:41:18Z

...nnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mongodb_source_to_assert.conf

        c_string = string
        c_boolean = boolean
-        c_tinyint = tinyint
-        c_smallint = smallint
+        c_tinyint = int


Why change it? Connector should support all type that SeaTunnel has defined, although mongo does not distinguish, you should and convert logic in your connector.

@TyrantLucifer You can take a look at the documentation of mongodb connector. The data type of mongodb and the data type of st are not fully mapped.

For example, source, write mongdb through the row of st, and the BSON of mongodb is not distinguished by TINYINT and SMALLINT.

TyrantLucifer · 2023-04-20T03:42:24Z

...ector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java


    private GenericContainer<?> mongodbContainer;
    private MongoClient client;

+    private static final Random random = new Random();


static parameter should on the top.

TyrantLucifer · 2023-04-20T03:43:15Z

...2-e2e/connector-mongodb-e2e/src/test/resources/mongodb_matchProjection_source_to_assert.conf

+    rules =
+      {
+        row_rules = [
+          {


Add more check rules.

hailin0 · 2023-04-23T07:30:38Z

...nnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mongodb_source_to_assert.conf

    database = "test_db"
-    collection = "source_table"
+    collection = "sink_table"
+    no-timeout = true


Unify the parameter naming style

hailin0 · 2023-04-23T07:32:18Z

...nnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mongodb_source_to_assert.conf

 }

 source {
-  MongoDB {
-    uri = "mongodb://e2e_mongodb:27017/test_db?retryWrites=true&writeConcern=majority"
+  Mongodb {


why change this name

MonsterChenzhuo · 2023-05-05T17:26:04Z

I tested 400w mongodb table reading today.

MonsterChenzhuo · 2023-05-06T02:34:56Z

100G 460w Line data. Mongodb record test EngineServer three nodes, 10 concurrent execution.

MonsterChenzhuo · 2023-05-06T03:39:30Z

Test writing 1000w pieces of data

docs/en/connector-v2/sink/MongoDB.md

docs/en/connector-v2/source/MongoDB.md

hailin0 · 2023-05-09T05:56:46Z

...src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSinkFactory.java

+                        MongodbConfig.COLLECTION,
+                        MongodbConfig.DATABASE,
+                        MongodbConfig.COLLECTION,


duplicate config COLLECTION ?

missing uri ?

...c/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/writer/MongodbWriter.java

...java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/config/MongodbWriterOptions.java

...main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/MongodbSourceFactory.java

...main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/reader/MongodbReader.java

MonsterChenzhuo · 2023-05-09T13:52:12Z

@hailin0 PTAL, thanks.
There is a problem with github in the afternoon. The code has not been fully pushed up. Now it is comprehensive. Please take a look.

hailin0 · 2023-05-09T15:33:40Z

...godb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java

+            doBulkWrite();
+        }
+    }
+


Suggested change

@Override

public Optional<Void> prepareCommit() {

doBulkWrite();

return Optional.empty();

}

@hailin0 Does your review mean what is described in the picture above?

hailin0 · 2023-05-09T15:34:09Z

...godb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java

+        }
+    }
+
+    void doBulkWrite() throws IOException {


Suggested change

void doBulkWrite() throws IOException {

synchronized void doBulkWrite() throws IOException {

MonsterChenzhuo · 2023-05-10T06:52:05Z

@hailin0 @EricJoy2048 PTAL, thanks.

EricJoy2048

LGTM， PTAL @hailin0 @TyrantLucifer

hailin0

LGTM

TyrantLucifer

Please provide multi split read e2e examples.

TyrantLucifer · 2023-05-12T02:07:19Z

...in/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/RowDataToBsonConverters.java

+            case ROW:
+                return createRowConverter((SeaTunnelRowType) type);
+            default:
+                throw new UnsupportedOperationException("Not support to parse type: " + type);


MongoDBConnectorException

TyrantLucifer · 2023-05-12T02:08:32Z

.../src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongoKeyExtractor.java

+
+    private static final long serialVersionUID = 1L;
+
+    private static String[] upsertKey;


TyrantLucifer · 2023-05-12T02:09:46Z

...ongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSink.java

+                            .withDatabase(database)
+                            .withCollection(collection)
+                            .withFlushSize(
+                                    pluginConfig.hasPath(MongodbConfig.BUFFER_FLUSH_MAX_ROWS.key())


use if-else instead

TyrantLucifer · 2023-05-12T02:10:55Z

...ongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSink.java

-        boolean useSimpleTextSchema = CatalogTableUtil.buildSimpleTextSchema().equals(rowType);
-        return new MongodbSinkWriter(rowType, useSimpleTextSchema, params);
+        return new MongodbWriter(
+                new RowDataDocumentSerializer(


add context into writer, it will be used to dirty records collect

TyrantLucifer · 2023-05-12T02:12:20Z

...db/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/MongodbSource.java

+                        rowType.getFieldNames(),
+                        rowType,
+                        pluginConfig.hasPath(MongodbConfig.FLAT_SYNC_STRING.key())
+                                ? pluginConfig.getBoolean(MongodbConfig.FLAT_SYNC_STRING.key())


The same as above

TyrantLucifer · 2023-05-12T02:15:04Z

...nnector-mongodb-e2e/src/test/resources/matchIT/mongodb_matchProjection_source_to_assert.conf

+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config


the same as above

TyrantLucifer · 2023-05-12T02:15:25Z

...2e/connector-mongodb-e2e/src/test/resources/matchIT/mongodb_matchQuery_source_to_assert.conf

+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config


the same as above

TyrantLucifer · 2023-05-12T02:15:56Z

...nector-mongodb-e2e/src/test/resources/updateIT/fake_source_to_updateMode_insert_mongodb.conf

+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config


the same as above

TyrantLucifer · 2023-05-12T02:16:09Z

...-v2-e2e/connector-mongodb-e2e/src/test/resources/updateIT/fake_source_to_update_mongodb.conf

+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config


the same as above

TyrantLucifer · 2023-05-12T02:16:36Z

...ector-v2-e2e/connector-mongodb-e2e/src/test/resources/updateIT/update_mongodb_to_assert.conf

+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######


the same as above

TyrantLucifer · 2023-05-12T02:19:01Z

...va/org/apache/seatunnel/connectors/seatunnel/mongodb/internal/MongodbCollectionProvider.java

+/** A builder class for creating {@link MongodbClientProvider}. */
+public class MongodbCollectionProvider {
+
+    public static Builder getBuilder() {


Suggested change

public static Builder getBuilder() {

public static Builder builder() {

TyrantLucifer · 2023-05-12T02:22:58Z

IMO, we will implement ddl change in the future. so the write logic of all connectors should not use async write, write action should bundle with checkpoint and preCommit stage cc @hailin0 @EricJoy2048

MonsterChenzhuo · 2023-05-15T12:25:54Z

@TyrantLucifer PTAL, thanks.

TyrantLucifer · 2023-05-15T15:18:29Z

...ector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java

@@ -67,36 +58,145 @@
 @Slf4j
 public class MongodbIT extends TestSuiteBase implements TestResource {

-    private static final String MONGODB_IMAGE = "mongo:6.0.5";
+    private static final Random random = new Random();


Suggested change

private static final Random random = new Random();

private static final Random RANDOM = new Random();

TyrantLucifer

LGTM

MonsterChenzhuo changed the title ~~[Feature][Connector-v2][Mongodb]Refactor mongodb source connector~~ [Feature][Connector-v2][Mongodb]Refactor mongodb connector Apr 19, 2023

TyrantLucifer requested changes Apr 19, 2023

View reviewed changes

TyrantLucifer reviewed Apr 20, 2023

View reviewed changes

MonsterChenzhuo force-pushed the mongodb2 branch 5 times, most recently from 8949a6f to c75e0ab Compare April 23, 2023 07:23

hailin0 reviewed Apr 23, 2023

View reviewed changes

MonsterChenzhuo force-pushed the mongodb2 branch 2 times, most recently from 0a39149 to 09ee942 Compare April 23, 2023 08:50

[Feature][Connector-v2][Mongodb]Refactor mongodb source

cd535dc

MonsterChenzhuo force-pushed the mongodb2 branch from 372c9e4 to cd535dc Compare April 23, 2023 10:29

Optimize e2e logic, optimize document logic

dc3c0ff

MonsterChenzhuo force-pushed the mongodb2 branch 3 times, most recently from ba17d01 to 5c98098 Compare April 27, 2023 13:45

Review problem fix and mongodb support transaction submission

b2f1f39

MonsterChenzhuo force-pushed the mongodb2 branch 2 times, most recently from b1f2c1d to 22df9ab Compare April 28, 2023 11:19

Fix the bug of submitting things and optimize the code logic

ca51c48

MonsterChenzhuo force-pushed the mongodb2 branch from f1be51b to ca51c48 Compare April 28, 2023 11:25

Merge branch 'dev' into mongodb2

ff55078

MonsterChenzhuo force-pushed the mongodb2 branch from e5b4977 to bc06ecf Compare April 28, 2023 12:57

MonsterChenzhuo force-pushed the mongodb2 branch from c6332e5 to b972922 Compare May 6, 2023 07:21

Fix code style

9152d04

MonsterChenzhuo force-pushed the mongodb2 branch from 5336ccf to 9152d04 Compare May 8, 2023 10:02

hailin0 reviewed May 9, 2023

View reviewed changes

Solve the problem of review

ed44399

MonsterChenzhuo force-pushed the mongodb2 branch from 24a3d44 to ed44399 Compare May 9, 2023 12:29

hailin0 reviewed May 9, 2023

View reviewed changes

MonsterChenzhuo force-pushed the mongodb2 branch 2 times, most recently from ac75c9c to 84bd528 Compare May 10, 2023 02:54

Solve the problem of review and optimize the document

e433ea2

MonsterChenzhuo force-pushed the mongodb2 branch from f29d155 to e433ea2 Compare May 10, 2023 06:22

EricJoy2048 reviewed May 10, 2023

View reviewed changes

hailin0 previously approved these changes May 12, 2023

View reviewed changes

TyrantLucifer requested changes May 12, 2023

View reviewed changes

TyrantLucifer reviewed May 12, 2023

View reviewed changes

MonsterChenzhuo dismissed hailin0’s stale review via d8602fc May 15, 2023 11:34

MonsterChenzhuo force-pushed the mongodb2 branch 2 times, most recently from 98d82d8 to 7d76fd0 Compare May 15, 2023 12:41

Solve the problem of review

2a09774

MonsterChenzhuo force-pushed the mongodb2 branch from e208d3d to 2a09774 Compare May 15, 2023 12:48

TyrantLucifer reviewed May 15, 2023

View reviewed changes

Solve the problem of review

e3c110b

TyrantLucifer approved these changes May 16, 2023

View reviewed changes

github-actions bot added approved reviewed labels May 16, 2023

hailin0 approved these changes May 16, 2023

View reviewed changes

hailin0 merged commit 5b1a843 into apache:dev May 16, 2023

		@@ -106,75 +83,6 @@ public void initConnection() {
		client = MongoClients.create(url);
		}

+    @Override
+    public Optional<Void> prepareCommit() {
+        doBulkWrite();
+        return Optional.empty();
+    }

	void doBulkWrite() throws IOException {
	synchronized void doBulkWrite() throws IOException {


		private static final long serialVersionUID = 1L;

		private static String[] upsertKey;

	public static Builder getBuilder() {
	public static Builder builder() {

	private static final Random random = new Random();
	private static final Random RANDOM = new Random();

[Feature][Connector-v2][Mongodb]Refactor mongodb connector #4620

[Feature][Connector-v2][Mongodb]Refactor mongodb connector #4620

Conversation

MonsterChenzhuo commented Apr 19, 2023 • edited Loading

Purpose of this pull request

Check list

MonsterChenzhuo commented Apr 19, 2023

MonsterChenzhuo commented Apr 19, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MonsterChenzhuo commented May 5, 2023

MonsterChenzhuo commented May 6, 2023

MonsterChenzhuo commented May 6, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MonsterChenzhuo commented May 9, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MonsterChenzhuo commented May 10, 2023

EricJoy2048 left a comment

Choose a reason for hiding this comment

hailin0 left a comment

Choose a reason for hiding this comment

TyrantLucifer left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

TyrantLucifer commented May 12, 2023 • edited Loading

MonsterChenzhuo commented May 15, 2023

Choose a reason for hiding this comment

TyrantLucifer left a comment

Choose a reason for hiding this comment

MonsterChenzhuo commented Apr 19, 2023 •

edited

Loading

MonsterChenzhuo commented May 9, 2023 •

edited

Loading

TyrantLucifer commented May 12, 2023 •

edited

Loading