You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Jiabao-Sun (via GitHub)" <gi...@apache.org> on 2023/02/14 08:29:35 UTC
[GitHub] [flink-connector-mongodb] Jiabao-Sun opened a new pull request, #3: [hotfix] Keep offset in split state to prevents duplicate records on restore.
Jiabao-Sun opened a new pull request, #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3
## What is the purpose of the change
Exact-once semantics may not be guaranteed at present on partial reads.
We use a number fetchSize to limit the records count for every fetch loop but we didn't record the offset into the split state. When resuming the split reader from a partially completed split, we may re-read some data.
In this PR, we record the current reading offset into split state.
Skip this offset when restoring to prevent duplicate reading.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] zentol commented on pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3#issuecomment-1433103256
We generally assume that the data in mongodb doesn't change for the lifetime of the job; otherwise you'd end up with serious inconsistency issues.
That's something that cdc connectors handle.
So to me it sounds we don't need the sorting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] zentol commented on a diff in pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3#discussion_r1105623716
##########
flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/source/enumerator/MongoSourceEnumStateSerializerTest.java:
##########
@@ -76,6 +76,7 @@ private static MongoScanSourceSplit createSourceSplit(int index) {
"coll",
new BsonDocument("_id", new BsonInt32(index)),
new BsonDocument("_id", MongoConstants.BSON_MAX_KEY),
- ID_HINT);
+ ID_HINT,
+ index);
Review Comment:
This change shouldnt be necessary.
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/split/MongoSourceSplitState.java:
##########
@@ -19,28 +19,19 @@
import org.apache.flink.annotation.PublicEvolving;
+import org.bson.BsonDocument;
+
/** MongoDB source split state. */
@PublicEvolving
-public class MongoSourceSplitState {
+public abstract class MongoSourceSplitState {
Review Comment:
I don't necessarily disagree with this change, just curious what brought it on. It doesn't seem to be required for fixing the issue at hand.
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/reader/split/MongoScanSourceSplitReader.java:
##########
@@ -181,6 +181,11 @@ private MongoCursor<BsonDocument> getOrCreateCursor() {
.hint(currentSplit.getHint())
.noCursorTimeout(readOptions.isNoCursorTimeout());
+ // Current split is partial read and recovered from checkpoint.
Review Comment:
```suggestion
// Current split was partially read and recovered from checkpoint
```
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/split/MongoSourceSplitState.java:
##########
@@ -19,28 +19,19 @@
import org.apache.flink.annotation.PublicEvolving;
+import org.bson.BsonDocument;
+
/** MongoDB source split state. */
@PublicEvolving
-public class MongoSourceSplitState {
+public abstract class MongoSourceSplitState {
- private final MongoSourceSplit split;
+ protected final MongoSourceSplit split;
public MongoSourceSplitState(MongoSourceSplit split) {
this.split = split;
}
- public MongoSourceSplit toMongoSourceSplit() {
- if (split instanceof MongoScanSourceSplit) {
- MongoScanSourceSplit scanSplit = (MongoScanSourceSplit) split;
- return new MongoScanSourceSplit(
- scanSplit.splitId(),
- scanSplit.getDatabase(),
- scanSplit.getCollection(),
- scanSplit.getMin(),
- scanSplit.getMax(),
- scanSplit.getHint());
- } else {
- throw new IllegalStateException("Unknown split type");
- }
- }
+ public abstract MongoSourceSplit toMongoSourceSplit();
+
+ public abstract void updateOffset(BsonDocument record);
Review Comment:
Not sure why we pass in the record; we're just incrementing the count. Will this be required in the future?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] zentol merged pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol merged PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] zentol commented on a diff in pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3#discussion_r1109995851
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/split/MongoSourceSplitState.java:
##########
@@ -19,28 +19,19 @@
import org.apache.flink.annotation.PublicEvolving;
+import org.bson.BsonDocument;
+
/** MongoDB source split state. */
@PublicEvolving
-public class MongoSourceSplitState {
+public abstract class MongoSourceSplitState {
Review Comment:
Maybe consider introducing a `MongoSourceSplitState` interface instead. We have then more freedom to change the internals than when we make the implementation itself part of the API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] zentol commented on a diff in pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3#discussion_r1111691646
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/split/MongoScanSourceSplitState.java:
##########
@@ -23,18 +23,19 @@
/** MongoDB source split state for {@link MongoScanSourceSplit}. */
@PublicEvolving
Review Comment:
this can now be `@Internal`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] Jiabao-Sun commented on a diff in pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3#discussion_r1108627598
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/split/MongoSourceSplitState.java:
##########
@@ -19,28 +19,19 @@
import org.apache.flink.annotation.PublicEvolving;
+import org.bson.BsonDocument;
+
/** MongoDB source split state. */
@PublicEvolving
-public class MongoSourceSplitState {
+public abstract class MongoSourceSplitState {
Review Comment:
Thanks @zentol for the review.
The main purpose of making these abstractions is to realize unbounded CDC reading in future by [Change Streams](https://www.mongodb.com/docs/manual/changeStreams/#change-streams) feature.
And the offset of the change stream split may be a `BsonDocument` type `ResumeToken` which can be extracted from a change record.
- The `updateOffset` method provides a change record parameter to extract ResumeToken.
- The offset has different data types so we may have some different `SplitState`.
But as you say, it probably doesn't need to be changed prematurely in order to fix this problem.
Do we need to remove these modifications?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] Jiabao-Sun commented on a diff in pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3#discussion_r1111733731
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/split/MongoScanSourceSplitState.java:
##########
@@ -23,18 +23,19 @@
/** MongoDB source split state for {@link MongoScanSourceSplit}. */
@PublicEvolving
Review Comment:
Changed MongoSourceSplitState and MongoScanSourceSplitState from PublicEvolving to Internal.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] Jiabao-Sun commented on pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3#issuecomment-1429801465
Hi @rozza.
Sorry to bother you. There's a question I want to ask you for advice.
Looking forward to your reply, when you have time.
In order to prevent the duplicate reading when restoring from a checkpoint, I keep the number of documents that have been read. When a failure occurs and retry, we can skip these documents and continue reading.
For this to work properly though MongoDB needs to return records in the exact same order for every query.
I want to confirm with you whether `hint()` will guarantee the same order of the returned data, or do we have to explicitly declare a `sort()` to ensure the same order.
```java
FindIterable<BsonDocument> findIterable =
mongoClient
.getDatabase(connectionOptions.getDatabase())
.getCollection(connectionOptions.getCollection(), BsonDocument.class)
.find()
// skip the documents has been read.
.skip(offset)
.min(currentSplit.getMin())
.max(currentSplit.getMax())
// hint _id or sharded key.
.hint(currentSplit.getHint());
// Is sort() required ?
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] rozza commented on pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "rozza (via GitHub)" <gi...@apache.org>.
rozza commented on PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3#issuecomment-1432780477
Hi @Jiabao-Sun,
As indexes are mutable relying on the natural sort order of them is not going to guarantee the same order of data - so sorting is required, it may end up being very cheap with the hint anyway.
Mutability is the heart of the issue - is the data static between the bounds of min and max for the duration of the processes? If not then you cannot guarantee that all data is covered and the offset would not be enough for a comparison, it may miss data if new data was added.
So for exactly-once semantics it requires the data to be static and no changes of data between the bounds for the duration of the processes.
I hope that helps,
Ross
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-mongodb] Jiabao-Sun commented on pull request #3: [FLINK-31063] Prevent duplicate reading when restoring from a checkpoint.
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #3:
URL: https://github.com/apache/flink-connector-mongodb/pull/3#issuecomment-1432841489
Thanks @rozza, That's very helpful for me.
When recovering from a short-term failure, the offset may help.But when we resume reading after a long time, the result may be inaccurate.
Generally speaking, batch reading is a one-time operation, and the checkpoint is used to recover immediately when a failure occurs. However, using offset may not be very precise, considering the data is non-static.
@zentol
Do we need to read split data at once to try to achieve exactly-once semantics?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org