Manas-Bhole opened a new issue, #111:
URL: https://github.com/apache/doris-kafka-connector/issues/111
**Versions**
- doris-kafka-connector 26.1.0 (same code on master, 40b80e9)
- Doris 4.1.5 (one FE, one BE), Kafka Connect 4.3.0, Debezium 3.6.3
(Postgres source)
- On Kafka 4.3 the connector needs commons-lang3 in its plugin directory
(#109); these runs had it
- Connector configured as in the "Synchronizing data captured by Debezium"
guide (`converter.mode=debezium_ingestion`, `enable.delete=true`,
`enable.combine.flush=true`, `buffer.flush.time=30`,
`buffer.count.records=10000`)
**What happens**
When an asynchronous stream load fails (BE restarting, connection refused, a
timeout), the records of that load are never written, the tasks keep running,
and the next commit returns offsets past them.
Sequence from the connector log (one task; the four failed loads ended at
offsets 7,177 to 7,481, and the next commit of each partition returned an
offset between 9,644 and 9,952):
```
06:43:44.977 stream load started for
pg_public_kv__KC_-1__KC_app_kv__KC_7481__KC_176
06:43:47.068 ERROR worker running error (Connect to 172.31.240.201:8040
failed: Connection refused)
06:43:47.068 LoadAsyncExecutor stop
06:43:47.410 Retrying to insert records to Doris, remaining retries: 9
06:43:47.410 LoadAsyncExecutor start
06:44:04.833 trigger flush by commit
06:44:04.914 load Result {...}
06:44:04.915 Returning committed offsets
{pg.public.kv-2=OffsetAndMetadata{offset=9952, ...}}
```
**Why**
1.
[`LoadAsyncExecutor.run()`](https://github.com/apache/doris-kafka-connector/blob/40b80e9897508baf56728b3757725947d103957c/src/main/java/org/apache/doris/kafka/connector/writer/load/AsyncDorisStreamLoad.java#L217-L223)
catches the failure, sets `exception`, clears `flushQueue` and exits. The
failed buffer is gone from memory.
2. The next `put()` calls `insert()`, which throws from
`checkFlushException()`.
[`DorisSinkTask.put()`](https://github.com/apache/doris-kafka-connector/blob/40b80e9897508baf56728b3757725947d103957c/src/main/java/org/apache/doris/kafka/connector/DorisSinkTask.java#L103-L121)
catches it, calls `sink.init()` (which restarts the executor and resets
`exception` to null) and throws `RetriableException`.
3. Kafka Connect re-delivers only the batch of that `put()`. The records of
the failed buffer came from earlier `put()` calls and are not re-delivered.
4. The next `preCommit()` flushes the new buffer and returns the latest
offsets from `topicPartitionOffset`, past the lost records.
Kafka Connect rewinds only if `preCommit()` sees the exception first
(`Offset commit failed, rewinding to last committed offsets`). Because `put()`
runs many times between commits, it usually sees it first.
**Reproduction**
Postgres inserts into new keys only (500 transactions/s for 240 s, 119,985
rows), so a lost record is a missing row. Faults on the Doris side only, 10 s
every 30 s: BE killed or paused, FE killed, and the FE HTTP connection cut,
delayed by 40 s, or truncated after 2 KB.
- 20 `worker running error`, each followed by `Retrying to insert records`;
only 4 commits failed and rewound
- tasks RUNNING throughout; at the end the committed offsets were the last
offset of every partition (29,982 / 29,744 / 30,205 / 30,051 for 29,983 /
29,745 / 30,206 / 30,052 messages)
- 35,140 of the 119,985 rows (29%) never reached Doris; the topic held no
duplicates
- two more runs with the same setup lost 46,169 and 46,690 rows (38% and 39%)
**Expected**
A failed load should not be forgotten: either keep the failed buffer and
send it again, or fail the commit (or the task) so that Kafka Connect rewinds
to the last committed offset.
**Suggested fix**
In `DorisSinkTask.put()`, when the exception comes from an earlier
asynchronous load, do not reset it and retry the current batch; make the next
`preCommit()` fail (or rewind with `context.offset()` to the last offset that
Doris confirmed), so that the records of the failed load are read again.
cc @JNSimba
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
