goutamadwant opened a new pull request, #13006:
URL: https://github.com/apache/seatunnel/pull/13006
### Purpose of this pull request
Closes #12565.
With `exactly_once = true`, restoring a CDC job from a savepoint after
adding a table rebuilds an incremental split that holds both the old tables (no
snapshot info) and the new table (with snapshot info).
`IncrementalSourceStreamFetcher.configureFilter()` only gave a table a starting
boundary when it had snapshot info, so every change of the old tables was
dropped after the restore.
This is a focused replacement for #12567 and follows the restart-position
review there:
- `IncrementalSplitAssigner` keeps the restart position as on dev (the
earliest offset of the restored splits). It also keeps each restored table's
own offset and passes it in the new `IncrementalSplit` as
`tableStartupOffsets`, only for tables whose offset is later than the split
start.
- `configureFilter()` starts every table without snapshot info from that
offset, or from the split start if there is none. A table that also has
snapshot info uses the later of the two.
- `IncrementalSplitState` keeps those offsets in the checkpoint until the
split has read past them, so restoring again in the middle of the replay keeps
them.
With `incremental.parallelism > 1`, a table that is regrouped into a split
starting earlier is emitted from its own offset instead of being replayed. A
record at exactly that offset is still emitted, the same rule as before,
because Oracle, MongoDB and DB2 offsets can be shared by several records and
skipping it could drop data.
### Does this PR introduce _any_ user-facing change?
Yes. After a savepoint restore that adds tables with `exactly_once = true`,
changes of the existing tables are no longer lost.
`IncrementalSplit` gets one new field. Checkpoints written before this
change restore as before (the field reads as empty), and older versions ignore
it.
### How was this patch tested?
- New unit tests in `IncrementalSourceStreamFetcherTest`,
`IncrementalSplitAssignerTest` and `IncrementalSplitStateTest`:
- the mixed split from the issue; this test fails on dev;
- two restored splits at different offsets regrouped into one: the earlier
table is not dropped, and the later table is not replayed before its own offset;
- a checkpoint taken during the replay, then restored again.
- New MySQL e2e
`testExactlyOnceRestoreWithAddedTableKeepsChangesOfExistingTables`:
1. savepoint;
2. add a table;
3. write to the old tables while the job is stopped and while the new
table is being snapshotted;
4. restore.
The job writes to the usual upsert sink and to an append-only copy without
a primary key, so both missing and duplicated rows would show. On dev the old
table ends at 109 of 129 rows; with this change all tables match.
The e2e uses one incremental split because restoring a MySQL savepoint
that holds more than one is rejected (`getRestoreTableStruct`), so the
`incremental.parallelism > 1` case is covered by the unit tests.
- connector-cdc-base and connector-cdc-mysql unit tests on JDK 11 and 17.
### Check list
* [ ] If any new Jar binary package adding in your PR, please add License
Notice according
[New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md)
* [ ] If necessary, please update the documentation to describe the new
feature. https://github.com/apache/seatunnel/tree/dev/docs
* [ ] If necessary, please update `incompatible-changes.md` to describe the
incompatibility caused by this PR.
* [ ] If you are contributing the connector code, please check that the
following files are updated:
1. Update
[plugin-mapping.properties](https://github.com/apache/seatunnel/blob/dev/plugin-mapping.properties)
and add new connector information in it
2. Update the pom file of
[seatunnel-dist](https://github.com/apache/seatunnel/blob/dev/seatunnel-dist/pom.xml)
3. Add ci label in
[label-scope-conf](https://github.com/apache/seatunnel/blob/dev/.github/workflows/labeler/label-scope-conf.yml)
4. Add e2e testcase in
[seatunnel-e2e](https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/)
5. Update connector
[plugin_config](https://github.com/apache/seatunnel/blob/dev/config/plugin_config)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
