sollhui commented on code in PR #67869:
URL: https://github.com/apache/doris/pull/67869#discussion_r4003410280
##########
be/src/load/routine_load/data_consumer.cpp:
##########
@@ -1015,16 +1022,15 @@ Status KinesisDataConsumer::_process_records(
continue;
}
- // Track the last sequence number for this shard
- _committed_sequence_numbers[shard_id] = record.GetSequenceNumber();
-
// Move record into shared_ptr to avoid expensive copy
- auto record_ptr =
std::make_shared<Aws::Kinesis::Model::Record>(std::move(record));
+ KinesisQueueItem item;
+ item.shard_id = shard_id;
+ item.record =
std::make_shared<Aws::Kinesis::Model::Record>(std::move(record));
- if (!queue->controlled_blocking_put(record_ptr,
-
config::blocking_queue_cv_wait_timeout_ms)) {
- // Queue shutdown
- return Status::InternalError("Queue shutdown during record
processing");
+ if (!queue->controlled_blocking_put(item,
config::blocking_queue_cv_wait_timeout_ms)) {
+ // The group may have reached a batch boundary while this consumer
was still draining
+ // a prefetched response. The appended prefix remains a valid
batch.
+ return Status::OK();
Review Comment:
[P2] Propagate queue shutdown so the whole consumer exits successfully
Returning OK here makes `group_consume()` treat an interrupted enqueue as a
fully processed response. If a consumer owns shards A, B and C, and the group
reaches its batch limit while the producer is enqueueing A's response, the
group shuts down the queue, cancels the consumer and waits in
`_thread_pool.join()`. This helper now returns OK, so `group_consume()`
continues its inner shard loop and issues `GetRecords` for B and C even when
`_cancelled` is already true; cancellation and the remaining time budget are
only checked in the outer loop.
Consequently, completing an otherwise valid batch waits for unnecessary AWS
calls, potentially exceeding the task/transaction timeout with many shards or
slow requests. A source error from one of those extra calls can also turn the
batch into a failure and roll back the appended prefix.
I reproduced this with the unchanged `group_consume()` and
`_process_records()` methods from this commit in an isolated C++ harness, with
AWS/queue infrastructure stubbed. When A's enqueue returns false with
cancellation already set, the current code calls `GetRecords` for A, B and C.
Restoring the original error return as a control stops after A. This is not a
full Doris/AWS integration test.
Please propagate an explicit queue-shutdown/stop indication from
`_process_records()` and have `group_consume()` return OK immediately for that
condition, preserving graceful batch completion. Checking cancellation before
each shard request would also help. The existing
`QueueShutdownDuringPrefetchIsGraceful` test only calls the helper, so please
add a multi-shard test that asserts no subsequent source requests occur after
shutdown.
--
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]
