wgzhao opened a new pull request, #13111:
URL: https://github.com/apache/seatunnel/pull/13111
Yes, I am contributing to Apache SeaTunnel.
### Purpose of this pull request
Fixes #13108.
A multi-table MySQL-CDC job can map several source tables to a single Doris
table. `MultiTableSink` creates one `DorisSinkWriter` per source table, but
every sub-writer derives the same stream-load label prefix from the sink table
path, the job id and `context.getIndexOfSubtask()`. In non-2PC mode
`LabelGenerator` only appends `System.currentTimeMillis()`, so two sub-writers
that generate a label in the same millisecond produce the exact same label.
This is easy to hit at checkpoint boundaries, where
`MultiTableSinkWriter.snapshotState` walks every sub-writer, and at higher
flush rates.
Doris rejects the second load with `LABEL_ALREADY_EXISTS`. `DorisSinkWriter`
treats any status other than `Success`/`Publish Timeout` as fatal, so the
pipeline fails. Restoring from the savepoint replays the same window and the
failure repeats. The two loads carry different data from different source
tables, so treating the duplicate label as success would silently drop data;
the labels must be distinct.
### What changed
`LabelGenerator` now appends a process-wide sequence to non-2PC labels:
```java
private static final AtomicLong LABEL_SEQUENCE = new AtomicLong();
return labelPrefix
+ "_"
+ System.currentTimeMillis()
+ "_"
+ Long.toString(LABEL_SEQUENCE.getAndIncrement(), 36);
```
- All sub-writers of one job share the connector class loader, so the
sequence is shared by every sub-sink of that job. The label prefix still
contains the job id and subtask index, so labels from different jobs and
subtasks remain distinct as before.
- 2PC labels are unchanged (`prefix + checkpointId`) because they are used
by checkpoint-id based `abortPreCommit`/recovery.
- The suffix is `_` plus a base-36 sequence, a few characters. It keeps the
label within Doris' 128-character limit for normal prefix lengths.
- No configuration keys, public APIs, SPI contracts or checkpoint
serialization change. The label is an internal stream-load identifier.
### Tests
Added `DorisCDCSinkIT#testMultiTableCDCSinkToSingleDorisTable`:
- Two source tables (`mysql_cdc_e2e_source_table` and
`mysql_cdc_e2e_source_table_2`) are matched by one `table-pattern` and both
write to `test.e2e_table_sink_multi`.
- The job uses `sink.enable-2pc = "false"` and a 1s checkpoint interval.
- The test inserts 100 rows into each source table after the job is running,
then asserts that all 204 rows (2 initial rows per source table plus the 200
inserted rows) reach the single Doris table and that the job is still `RUNNING`.
- At each checkpoint both sub-writers generate labels in the same
millisecond. Without the fix the second load fails with `LABEL_ALREADY_EXISTS`,
the job stops and the row count cannot reach the expected value.
Verification performed locally:
- `./mvnw -pl
seatunnel-connectors-v2/connector-doris,seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e
spotless:apply`
- `./mvnw -pl seatunnel-connectors-v2/connector-doris -am -DskipTests
-Dskip.spotless=true -Dcheckstyle.skip=true -Drat.skip=true package`
- `./mvnw -pl seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e
-am -DskipTests -Dskip.spotless=true -Dcheckstyle.skip=true -Drat.skip=true
package`
The container-based E2E test is exercised by the E2E CI job; it needs the
Doris and MySQL test containers.
Related reports with the same symptom without a root cause: #7050 and #9048.
### Compatibility
Configuration keys, defaults, public APIs/SPIs and checkpoint state
serialization are unchanged. Only the generated non-2PC stream-load label shape
changes; 2PC labels stay deterministic.
--
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]
