goutamadwant opened a new pull request, #12848:
URL: https://github.com/apache/seatunnel/pull/12848
### Purpose of this pull request
Fixes #12847.
In CDC mode the Databend sink writes rows to a raw table and moves them to
the target table with `MERGE INTO`. On Flink 1.15–1.18 the aggregated committer
runs in a separate global committer task that drops commits arriving after end
of data and is never closed, so if no checkpoint completes after the last
insert, the `MERGE` never runs. `DatabendCDCSinkIT` fails this way on Flink
1.18 (`expected: <3> but was: <0>`), and Flink 1.15 was disabled for the same
symptom in #11624.
`DatabendSinkWriter.close()` already tries a final `MERGE`, but
`performMerge()` checks `batchCount`, which `executeBatch()` has just reset to
0, so it never runs.
This PR tracks CDC rows written to the raw table but not yet merged with a
flag, so the writer's final `MERGE` runs when there is something to merge.
The writer and the aggregated committer may now both merge the same stream.
That is safe: when several `MERGE`s read one stream at the same time, one
applies the rows and the others merge nothing (tested below).
**Before:** on Flink 1.15–1.18 the job finishes with the rows in the raw
table and an empty target table.
**After:** the writer merges the remaining rows when it closes, on every
engine.
### Does this PR introduce _any_ user-facing change?
Bug fix only: on Flink 1.15–1.18, CDC rows written after the last completed
checkpoint now reach the target table. No config or API change.
### How was this patch tested?
- `DatabendSinkWriterTest`: added tests that `close()` merges after a CDC
batch was flushed and skips the merge when nothing was written. The first fails
without the fix. 15/15 pass on JDK 11 and 17.
- `DatabendCDCSinkIT` with `checkpoint.interval = 30000`, so no checkpoint
completes before the writes (deterministic repro):
- before (`dev`): fails on Flink 1.18 with `expected: <3> but was: <0>`;
Flink 1.20, Spark 3.3 and Zeta pass;
- after: passes on Flink 1.18, Flink 1.20, Spark 3.3 and Zeta.
- `DatabendCDCSinkIT` with the default `checkpoint.interval = 1000`: passes
on all four engines.
- Concurrent merges, on the `datafuselabs/databend:nightly` image used by
the IT: 9 simultaneous `MERGE`s from one stream, and 8 writers inserting and
merging while another session merged every 200 ms. No errors, no duplicate
rows, and every row reached the target table.
### 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]
