This is an automated email from the ASF dual-hosted git repository.
liugddx pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 526197af67 [Hotfix][Connector-V2][Mongodb] Fix document error content
and remove redundant code (#4982)
526197af67 is described below
commit 526197af67557676cd6008f42d1ccdc4c58786aa
Author: monster <[email protected]>
AuthorDate: Wed Jun 28 11:15:45 2023 +0800
[Hotfix][Connector-V2][Mongodb] Fix document error content and remove
redundant code (#4982)
Co-authored-by: chenzy15 <[email protected]>
---
docs/en/connector-v2/sink/MongoDB.md | 55 ++++++++++------------
docs/en/connector-v2/source/MongoDB.md | 40 +++++++---------
.../seatunnel/mongodb/config/MongodbConfig.java | 6 +--
.../mongodb/serde/DocumentRowDataDeserializer.java | 4 +-
.../mongodb/serde/RowDataDocumentSerializer.java | 2 +-
.../seatunnel/mongodb/sink/MongodbSink.java | 4 +-
.../seatunnel/mongodb/sink/MongodbWriter.java | 4 +-
.../seatunnel/mongodb/source/MongodbSource.java | 8 ++--
.../source/enumerator/MongodbSplitEnumerator.java | 9 ++--
.../mongodb/source/reader/MongodbReader.java | 9 ++--
10 files changed, 60 insertions(+), 81 deletions(-)
diff --git a/docs/en/connector-v2/sink/MongoDB.md
b/docs/en/connector-v2/sink/MongoDB.md
index 41cd463c63..40355583cb 100644
--- a/docs/en/connector-v2/sink/MongoDB.md
+++ b/docs/en/connector-v2/sink/MongoDB.md
@@ -2,8 +2,7 @@
> MongoDB Sink Connector
-Support Those Engines
----------------------
+## Support Those Engines
> Spark<br/>
> Flink<br/>
@@ -19,14 +18,12 @@ Key Features
> 1.If you want to use CDC-written features, recommend enable the
> upsert-enable configuration.
-Description
------------
+## Description
The MongoDB Connector provides the ability to read and write data from and to
MongoDB.
This document describes how to set up the MongoDB connector to run data
writers against MongoDB.
-Supported DataSource Info
--------------------------
+## Supported DataSource Info
In order to use the Mongodb connector, the following dependencies are required.
They can be downloaded via install-plugin.sh or from the Maven central
repository.
@@ -35,8 +32,7 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
|------------|--------------------|---------------------------------------------------------------------------------------------------------------|
| MongoDB | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/seatunnel-connectors-v2/connector-mongodb)
|
-Data Type Mapping
------------------
+## Data Type Mapping
The following table lists the field data type mapping from MongoDB BSON type
to Seatunnel data type.
@@ -63,21 +59,20 @@ The following table lists the field data type mapping from
MongoDB BSON type to
> 1.When using SeaTunnel to write Date and Timestamp types to MongoDB, both
> will produce a Date data type in MongoDB, but the precision will be
> different. The data generated by the SeaTunnel Date type has second-level
> precision, while the data generated by the SeaTunnel Timestamp type has
> millisecond-level precision.<br/>
> 2.When using the DECIMAL type in SeaTunnel, be aware that the maximum range
> cannot exceed 34 digits, which means you should use decimal(34, 18).<br/>
-Sink Options
-------------
-
-| Name | Type | Required | Default |
Description |
-|-----------------------|----------|----------|---------|---------------------------------------------------------------------------------------------------|
-| uri | String | Yes | - | The MongoDB
connection uri.
|
-| database | String | Yes | - | The name of MongoDB
database to read or write. |
-| collection | String | Yes | - | The name of MongoDB
collection to read or write. |
-| schema | String | Yes | - | MongoDB's BSON and
seatunnel data structure mapping |
-| buffer-flush.max-rows | String | No | 1000 | Specifies the
maximum number of buffered rows per batch request.
|
-| buffer-flush.interval | String | No | 30000 | Specifies the retry
time interval if writing records to database failed, the unit is seconds. |
-| retry.max | String | No | 3 | Specifies the max
retry times if writing records to database failed.
|
-| retry.interval | Duration | No | 1000 | Specifies the retry
time interval if writing records to database failed, the unit is millisecond. |
-| upsert-enable | Boolean | No | false | Whether to write
documents via upsert mode.
|
-| primary-key | List | No | - | The primary keys for
upsert/update. Keys are in `["id","name",...]` format for properties. |
+## Sink Options
+
+| Name | Type | Required | Default |
Description
|
+|-----------------------|----------|----------|---------|------------------------------------------------------------------------------------------------------------------------------|
+| uri | String | Yes | - | The MongoDB standard
connection uri. eg.
mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true.
|
+| database | String | Yes | - | The name of MongoDB
database to read or write.
|
+| collection | String | Yes | - | The name of MongoDB
collection to read or write.
|
+| schema | String | Yes | - | MongoDB's BSON and
seatunnel data structure mapping.
|
+| buffer-flush.max-rows | String | No | 1000 | Specifies the
maximum number of buffered rows per batch request.
|
+| buffer-flush.interval | String | No | 30000 | Specifies the
maximum interval of buffered rows per batch request, the unit is millisecond.
|
+| retry.max | String | No | 3 | Specifies the max
number of retry if writing records to database failed.
|
+| retry.interval | Duration | No | 1000 | Specifies the retry
time interval if writing records to database failed, the unit is millisecond.
|
+| upsert-enable | Boolean | No | false | Whether to write
documents via upsert mode.
|
+| primary-key | List | No | - | The primary keys for
upsert/update. Keys are in `["id","name",...]` format for properties.
|
**Tips**
@@ -85,8 +80,7 @@ Sink Options
> 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
--------------------------------------------------
+## How to Create a MongoDB Data Synchronization Jobs
The following example demonstrates how to create a data synchronization job
that writes randomly generated data to a MongoDB database:
@@ -128,10 +122,9 @@ sink {
}
```
-Parameter Interpretation
-------------------------
+## Parameter Interpretation
-**MongoDB Database Connection URI Examples**
+### MongoDB Database Connection URI Examples
Unauthenticated single node connection:
@@ -171,7 +164,7 @@
mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb
Note: The username and password in the URI must be URL-encoded before being
concatenated into the connection string.
-**Buffer Flush**
+### Buffer Flush
```bash
sink {
@@ -192,14 +185,14 @@ sink {
}
```
-**Why is Not Recommended to Use Transactions for Operation?**
+### Why is Not Recommended to Use Transactions for Operation?
Although MongoDB has fully supported multi-document transactions since version
4.2, it doesn't mean that everyone should use them recklessly.
Transactions are equivalent to locks, node coordination, additional overhead,
and performance impact.
Instead, the principle for using transactions should be: avoid using them if
possible.
The necessity for using transactions can be greatly avoided by designing
systems rationally.
-**Idempotent Writes**
+### Idempotent Writes
By specifying a clear primary key and using the upsert method, exactly-once
write semantics can be achieved.
diff --git a/docs/en/connector-v2/source/MongoDB.md
b/docs/en/connector-v2/source/MongoDB.md
index da9a19d0ec..14f283afb4 100644
--- a/docs/en/connector-v2/source/MongoDB.md
+++ b/docs/en/connector-v2/source/MongoDB.md
@@ -2,15 +2,13 @@
> MongoDB Source Connector
-Support Those Engines
----------------------
+## Support Those Engines
> Spark<br/>
> Flink<br/>
> SeaTunnel Zeta<br/>
-Key Features
-------------
+## Key Features
- [x] [batch](../../concept/connector-v2-features.md)
- [ ] [stream](../../concept/connector-v2-features.md)
@@ -19,14 +17,12 @@ Key Features
- [x] [parallelism](../../concept/connector-v2-features.md)
- [x] [support user-defined split](../../concept/connector-v2-features.md)
-Description
------------
+## Description
The MongoDB Connector provides the ability to read and write data from and to
MongoDB.
This document describes how to set up the MongoDB connector to run data reads
against MongoDB.
-Supported DataSource Info
--------------------------
+## Supported DataSource Info
In order to use the Mongodb connector, the following dependencies are required.
They can be downloaded via install-plugin.sh or from the Maven central
repository.
@@ -35,8 +31,7 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
|------------|--------------------|---------------------------------------------------------------------------------------------------------------|
| MongoDB | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/seatunnel-connectors-v2/connector-mongodb)
|
-Data Type Mapping
------------------
+## Data Type Mapping
The following table lists the field data type mapping from MongoDB BSON type
to SeaTunnel data type.
@@ -68,17 +63,16 @@ For specific types in MongoDB, we use Extended JSON format
to map them to SeaTun
> 1.When using the DECIMAL type in SeaTunnel, be aware that the maximum range
> cannot exceed 34 digits, which means you should use decimal(34, 18).<br/>
-Source Options
---------------
+## Source Options
| Name | Type | Required | Default |
Description
|
|----------------------|---------|----------|------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| uri | String | Yes | - | The MongoDB
connection uri.
|
+| uri | String | Yes | - | The MongoDB
standard connection uri. eg.
mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true.
|
| database | String | Yes | - | The name of
MongoDB database to read or write.
|
| collection | String | Yes | - | The name of
MongoDB collection to read or write.
|
-| schema | String | Yes | - | MongoDB's
BSON and seatunnel data structure mapping
|
+| schema | String | Yes | - | MongoDB's
BSON and seatunnel data structure mapping.
|
| match.query | String | No | - | In MongoDB,
filters are used to filter documents for query operations.
|
-| match.projection | String | No | - | In MongoDB,
Projection is used to control the fields contained in the query results
|
+| match.projection | String | No | - | In MongoDB,
Projection is used to control the fields contained in the query results.
|
| partition.split-key | String | No | _id | The key of
Mongodb fragmentation.
|
| partition.split-size | Long | No | 64 * 1024 * 1024 | The size of
Mongodb fragment.
|
| cursor.no-timeout | Boolean | No | true | MongoDB
server normally times out idle cursors after an inactivity period (10 minutes)
to prevent excess memory use. Set this option to true to prevent that. However,
if the application takes longer than 30 minutes to process the current batch of
documents, the session is marked as expired and closed. |
@@ -86,8 +80,7 @@ Source Options
| max.time-min | Long | No | 600 | This
parameter is a MongoDB query option that limits the maximum execution time for
query operations. The value of maxTimeMin is in Minute. If the execution time
of the query exceeds the specified time limit, MongoDB will terminate the
operation and return an error. |
| flat.sync-string | Boolean | No | true | By utilizing
flatSyncString, only one field attribute value can be set, and the field type
must be a String. This operation will perform a string mapping on a single
MongoDB data entry.
|
-How to Create a MongoDB Data Synchronization Jobs
--------------------------------------------------
+## How to Create a MongoDB Data Synchronization Jobs
The following example demonstrates how to create a data synchronization job
that reads data from MongoDB and prints it on the local client:
@@ -143,10 +136,9 @@ sink {
}
```
-Parameter Interpretation
-------------------------
+## Parameter Interpretation
-**MongoDB Database Connection URI Examples**
+### MongoDB Database Connection URI Examples
Unauthenticated single node connection:
@@ -186,7 +178,7 @@
mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb
Note: The username and password in the URI must be URL-encoded before being
concatenated into the connection string.
-**MatchQuery Scan**
+### MatchQuery Scan
In data synchronization scenarios, the matchQuery approach needs to be used
early to reduce the number of documents that need to be processed by subsequent
operators, thus improving performance.
Here is a simple example of a seatunnel using `match.query`
@@ -225,7 +217,7 @@ The following are examples of MatchQuery query statements
of various data types:
Please refer to how to write the syntax of
`match.query`：https://www.mongodb.com/docs/manual/tutorial/query-documents
-**Projection Scan**
+### Projection Scan
In MongoDB, Projection is used to control which fields are included in the
query results. This can be accomplished by specifying which fields need to be
returned and which fields do not.
In the find() method, a projection object can be passed as a second argument.
The key of the projection object indicates the fields to include or exclude,
and a value of 1 indicates inclusion and 0 indicates exclusion.
@@ -256,7 +248,7 @@ source {
```
-**Partitioned Scan**
+### Partitioned Scan
To speed up reading data in parallel source task instances, seatunnel provides
a partitioned scan feature for MongoDB collections. The following partitioning
strategies are provided.
Users can control data sharding by setting the partition.split-key for
sharding keys and partition.split-size for sharding size.
@@ -280,7 +272,7 @@ source {
```
-**Flat Sync String**
+### Flat Sync String
By utilizing `flat.sync-string`, only one field attribute value can be set,
and the field type must be a String.
This operation will perform a string mapping on a single MongoDB data entry.
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/config/MongodbConfig.java
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/config/MongodbConfig.java
index 6d008d5bd3..1ba9ad70cc 100644
---
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/config/MongodbConfig.java
+++
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/config/MongodbConfig.java
@@ -118,16 +118,16 @@ public class MongodbConfig {
public static final Option<Long> BUFFER_FLUSH_INTERVAL =
Options.key("buffer-flush.interval")
.longType()
- .defaultValue(30_000L)
+ .defaultValue(30000L)
.withDescription(
- "Specifies the retry time interval if writing
records to database failed.");
+ "Specifies the maximum interval of buffered rows
per batch request, the unit is millisecond.");
public static final Option<Integer> RETRY_MAX =
Options.key("retry.max")
.intType()
.defaultValue(3)
.withDescription(
- "Specifies the max retry times if writing records
to database failed.");
+ "Specifies the max number of retry if writing
records to database failed.");
public static final Option<Long> RETRY_INTERVAL =
Options.key("retry.interval")
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/DocumentRowDataDeserializer.java
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/DocumentRowDataDeserializer.java
index 15a0147ad9..3c00ae32b6 100644
---
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/DocumentRowDataDeserializer.java
+++
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/DocumentRowDataDeserializer.java
@@ -37,10 +37,10 @@ public class DocumentRowDataDeserializer implements
DocumentDeserializer<SeaTunn
private final BsonToRowDataConverters bsonConverters;
- private final Boolean flatSyncString;
+ private final boolean flatSyncString;
public DocumentRowDataDeserializer(
- String[] fieldNames, SeaTunnelDataType<?> dataTypes, Boolean
flatSyncString) {
+ String[] fieldNames, SeaTunnelDataType<?> dataTypes, boolean
flatSyncString) {
if (fieldNames == null || fieldNames.length < 1) {
throw new MongodbConnectorException(ILLEGAL_ARGUMENT, "fieldName
is empty");
}
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/RowDataDocumentSerializer.java
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/RowDataDocumentSerializer.java
index 2e320e4f45..26f2abbf40 100644
---
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/RowDataDocumentSerializer.java
+++
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/RowDataDocumentSerializer.java
@@ -43,7 +43,7 @@ import static
org.apache.seatunnel.common.exception.CommonErrorCode.ILLEGAL_ARGU
public class RowDataDocumentSerializer implements
DocumentSerializer<SeaTunnelRow> {
private final RowDataToBsonConverters.RowDataToBsonConverter
rowDataToBsonConverter;
- private final Boolean isUpsertEnable;
+ private final boolean isUpsertEnable;
private final Function<BsonDocument, BsonDocument> filterConditions;
private final Map<RowKind, WriteModelSupplier> writeModelSuppliers;
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSink.java
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSink.java
index f0c6a4c746..fa2c212c3d 100644
---
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSink.java
+++
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSink.java
@@ -33,7 +33,6 @@ import
org.apache.seatunnel.connectors.seatunnel.mongodb.serde.RowDataToBsonConv
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;
@@ -110,8 +109,7 @@ public class MongodbSink extends
AbstractSimpleSink<SeaTunnelRow, Void> {
}
@Override
- public AbstractSinkWriter<SeaTunnelRow, Void>
createWriter(SinkWriter.Context context)
- throws IOException {
+ public AbstractSinkWriter<SeaTunnelRow, Void>
createWriter(SinkWriter.Context context) {
return new MongodbWriter(
new RowDataDocumentSerializer(
RowDataToBsonConverters.createConverter(seaTunnelRowType),
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java
index 0361ad0606..794c3bf04a 100644
---
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java
+++
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java
@@ -33,7 +33,6 @@ import com.mongodb.client.model.BulkWriteOptions;
import com.mongodb.client.model.WriteModel;
import lombok.extern.slf4j.Slf4j;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
@@ -61,6 +60,7 @@ public class MongodbWriter extends
AbstractSinkWriter<SeaTunnelRow, Void> {
private volatile long lastSendTime = 0L;
+ // TODO：Reserve parameters.
private final SinkWriter.Context context;
public MongodbWriter(
@@ -103,7 +103,7 @@ public class MongodbWriter extends
AbstractSinkWriter<SeaTunnelRow, Void> {
}
@Override
- public void close() throws IOException {
+ public void close() {
doBulkWrite();
if (collectionProvider != null) {
collectionProvider.close();
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/MongodbSource.java
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/MongodbSource.java
index 4165a76b3a..e68796faa9 100644
---
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/MongodbSource.java
+++
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/MongodbSource.java
@@ -152,22 +152,20 @@ public class MongodbSource
}
@Override
- public SourceReader<SeaTunnelRow, MongoSplit>
createReader(SourceReader.Context readerContext)
- throws Exception {
+ public SourceReader<SeaTunnelRow, MongoSplit>
createReader(SourceReader.Context readerContext) {
return new MongodbReader(readerContext, clientProvider, deserializer,
mongodbReadOptions);
}
@Override
public SourceSplitEnumerator<MongoSplit, ArrayList<MongoSplit>>
createEnumerator(
- SourceSplitEnumerator.Context<MongoSplit> enumeratorContext)
throws Exception {
+ SourceSplitEnumerator.Context<MongoSplit> enumeratorContext) {
return new MongodbSplitEnumerator(enumeratorContext, clientProvider,
splitStrategy);
}
@Override
public SourceSplitEnumerator<MongoSplit, ArrayList<MongoSplit>>
restoreEnumerator(
SourceSplitEnumerator.Context<MongoSplit> enumeratorContext,
- ArrayList<MongoSplit> checkpointState)
- throws Exception {
+ ArrayList<MongoSplit> checkpointState) {
return new MongodbSplitEnumerator(
enumeratorContext, clientProvider, splitStrategy,
checkpointState);
}
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/enumerator/MongodbSplitEnumerator.java
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/enumerator/MongodbSplitEnumerator.java
index 466e853da4..206ea76889 100644
---
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/enumerator/MongodbSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/enumerator/MongodbSplitEnumerator.java
@@ -29,7 +29,6 @@ import
org.apache.seatunnel.connectors.seatunnel.mongodb.source.split.MongoSplit
import com.mongodb.MongoNamespace;
import lombok.extern.slf4j.Slf4j;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -73,7 +72,7 @@ public class MongodbSplitEnumerator
public void open() {}
@Override
- public synchronized void run() throws Exception {
+ public synchronized void run() {
log.info("Starting MongoSplitEnumerator.");
Set<Integer> readers = context.registeredReaders();
pendingSplits.addAll(strategy.split());
@@ -86,7 +85,7 @@ public class MongodbSplitEnumerator
}
@Override
- public void close() throws IOException {
+ public void close() {
if (clientProvider != null) {
clientProvider.close();
}
@@ -121,12 +120,12 @@ public class MongodbSplitEnumerator
}
@Override
- public ArrayList<MongoSplit> snapshotState(long checkpointId) throws
Exception {
+ public ArrayList<MongoSplit> snapshotState(long checkpointId) {
return pendingSplits;
}
@Override
- public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ public void notifyCheckpointComplete(long checkpointId) {
// Do nothing
}
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/reader/MongodbReader.java
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/reader/MongodbReader.java
index 464aa1b783..48ff273eaa 100644
---
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/reader/MongodbReader.java
+++
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/reader/MongodbReader.java
@@ -33,7 +33,6 @@ import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCursor;
import lombok.extern.slf4j.Slf4j;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
@@ -78,14 +77,14 @@ public class MongodbReader implements
SourceReader<SeaTunnelRow, MongoSplit> {
}
@Override
- public void close() throws IOException {
+ public void close() {
if (cursor != null) {
cursor.close();
}
}
@Override
- public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
+ public void pollNext(Collector<SeaTunnelRow> output) {
synchronized (output.getCheckpointLock()) {
MongoSplit currentSplit = pendingSplits.poll();
if (null != currentSplit) {
@@ -118,7 +117,7 @@ public class MongodbReader implements
SourceReader<SeaTunnelRow, MongoSplit> {
}
@Override
- public List<MongoSplit> snapshotState(long checkpointId) throws Exception {
+ public List<MongoSplit> snapshotState(long checkpointId) {
return new ArrayList<>(pendingSplits);
}
@@ -135,7 +134,7 @@ public class MongodbReader implements
SourceReader<SeaTunnelRow, MongoSplit> {
}
@Override
- public void notifyCheckpointComplete(long checkpointId) throws Exception {}
+ public void notifyCheckpointComplete(long checkpointId) {}
private void closeCurrentSplit() {
Preconditions.checkNotNull(cursor);
