github-actions[bot] commented on code in PR #65651:
URL: https://github.com/apache/doris/pull/65651#discussion_r3592531985
##########
be/src/load/stream_load/stream_load_recorder_manager.cpp:
##########
@@ -232,30 +234,45 @@ std::string
StreamLoadRecorderManager::_parse_and_format_record(const std::strin
void StreamLoadRecorderManager::_load_if_necessary() {
int64_t current_time = UnixMillis();
- bool should_load = _buffer.size() >=
config::stream_load_record_batch_bytes ||
+ bool should_load = _has_pending_batch() ||
Review Comment:
[P1] Back off retained-batch retries
Once `_pending_label` is set, this condition retries on every one-second
worker tick regardless of `_last_load_time`. Each fast failure rebuilds
`_buffer.ToString()` (a full copy of a batch whose default trigger is 100 MiB),
submits it through local HTTP/FE again, and emits another warning. A persistent
application failure such as an incompatible audit-table schema can therefore
make every BE copy/upload roughly a full batch per second and amplify the
outage. Keep the frozen payload, but track a next-retry time with capped
backoff/jitter and rate-limit the repeated error; reset that state after
acceptance.
##########
be/src/load/stream_load/stream_load_recorder_manager.cpp:
##########
@@ -287,17 +304,33 @@ Status StreamLoadRecorderManager::_send_stream_load(const
std::string& data) {
return Status::InternalError("Stream load failed with HTTP status {}:
{}", http_status,
response);
}
+ return _parse_stream_load_response(response);
+}
+
+Status StreamLoadRecorderManager::_parse_stream_load_response(const
std::string& response) {
rapidjson::Document doc;
- if (!doc.Parse(response.data(), response.length()).HasParseError()) {
- if (doc.HasMember("Status") && doc["Status"].IsString()) {
- std::string status = doc["Status"].GetString();
- if (status != "Success") {
- return Status::InternalError("Stream load status is not
Success: {}", response);
- }
- }
+ if (doc.Parse(response.data(), response.length()).HasParseError() ||
!doc.IsObject()) {
+ return Status::InternalError("Failed to parse stream load response:
{}", response);
+ }
+ if (!doc.HasMember("Status") || !doc["Status"].IsString()) {
+ return Status::InternalError("Stream load response has no valid
Status: {}", response);
+ }
+
+ std::string status = doc["Status"].GetString();
+ // PUBLISH_TIMEOUT means the transaction was committed but did not become
visible before the
+ // response timeout. The stream load path does not roll it back.
+ if (status == "Success" || status == "Publish Timeout") {
+ return Status::OK();
+ }
+ // A retry after an unknown HTTP outcome reuses the pending label.
FINISHED means the original
+ // transaction is already COMMITTED or VISIBLE, so the batch must not be
submitted again.
+ if (status == "Label Already Exists" && doc.HasMember("ExistingJobStatus")
&&
+ doc["ExistingJobStatus"].IsString() &&
+ std::string(doc["ExistingJobStatus"].GetString()) == "FINISHED") {
Review Comment:
[P1] Make the retry label unique to this BE and batch
`FINISHED` only proves that some transaction with this database-wide label
committed; it does not prove that it contains this payload. Every BE loads its
local recorder into the same `__internal_schema` database, while
`_generate_label()` contains only wall-clock time to millisecond precision. If
BEs A and B generate the same label, A can commit first and B will receive A's
`FINISHED`; this branch then returns OK, saves B's cursor, and clears B's rows
even though B's body was never accepted. Use a globally unique, durable batch
identity (for example a persistent BE identity plus a cursor-range/batch ID or
digest) before treating an existing finished label as acceptance, and cover two
managers producing the same timestamp.
##########
be/src/load/stream_load/stream_load_recorder_manager.cpp:
##########
@@ -321,6 +354,7 @@ void StreamLoadRecorderManager::_reset_batch(int64_t
current_time) {
_buffer.clear();
_last_load_time = current_time;
_record_num = 0;
+ _pending_label.clear();
Review Comment:
[P1] Keep the accepted batch until the cursor write succeeds
`_save_last_fetch_key()` returns `void` and only logs when its RocksDB
`put()` fails, but `_load_if_necessary()` then unconditionally reaches this
reset and clears both the payload and its stable label. If the remote audit
transaction committed under L but the cursor write fails, a restart loads the
old cursor, re-fetches the same rows, and submits them under a new label,
duplicating the committed batch. Propagate the cursor-write `Status` and
preserve the accepted batch/label until that cursor update succeeds (with a
state that retries persistence without resubmitting the payload); add a
failing-recorder/restart test.
##########
be/src/load/stream_load/stream_load_recorder_manager.cpp:
##########
@@ -89,7 +89,9 @@ void StreamLoadRecorderManager::stop() {
void StreamLoadRecorderManager::_worker_thread_func() {
SCOPED_ATTACH_TASK(_mem_tracker);
while (!_stop) {
- _fetch_and_buffer_records();
+ if (!_has_pending_batch()) {
Review Comment:
[P1] Handle an expired cursor key before freezing fetches
A pending failure can now keep `_last_fetch_key` unchanged longer than the
recorder's 8-hour TTL. Once compaction removes that key,
`StreamLoadRecorder::get_batch()` calls `Seek(start)`—which already lands on
the first greater surviving record—and then unconditionally calls `Next()`,
skipping that record. After this pending batch eventually succeeds and saves
the stale key, the next fetch therefore loses the first surviving audit entry.
Make `get_batch()` advance only when the seek result exactly equals `start`
(otherwise include the greater key), and add a missing/expired-cursor resume
test.
--
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]
