Skip to content

[fix][routine-load] fix bug that routine load cannot cancel task when append_data return error #8457

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
caiconghui merged 1 commit into from
Mar 14, 2022
Merged

[fix][routine-load] fix bug that routine load cannot cancel task when append_data return error #8457

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions be/src/runtime/routine_load/data_consumer_group.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,6 @@ Status KafkaDataConsumerGroup::start_all(StreamLoadContext* ctx) {

MonotonicStopWatch watch;
watch.start();
Status st;
bool eos = false;
while (true) {
if (eos || left_time <= 0 || left_rows <= 0 || left_bytes <= 0) {
Expand All @@ -140,12 +139,10 @@ Status KafkaDataConsumerGroup::start_all(StreamLoadContext* ctx) {
// waiting all threads finished
_thread_pool.shutdown();
_thread_pool.join();

if (!result_st.ok()) {
// some of consumers encounter errors, cancel this task
kafka_pipe->cancel(result_st.get_error_msg());
return result_st;
}

kafka_pipe->finish();
ctx->kafka_info->cmt_offset = std::move(cmt_offset);
ctx->receive_bytes = ctx->max_batch_size - left_bytes;
Expand All @@ -159,9 +156,8 @@ Status KafkaDataConsumerGroup::start_all(StreamLoadContext* ctx) {
<< ", partition: " << msg->partition() << ", offset: " << msg->offset()
<< ", len: " << msg->len();

(kafka_pipe.get()->*append_data)(static_cast<const char*>(msg->payload()),
Status st = (kafka_pipe.get()->*append_data)(static_cast<const char*>(msg->payload()),
static_cast<size_t>(msg->len()));

if (st.ok()) {
left_rows--;
left_bytes -= msg->len();
Expand All @@ -172,6 +168,12 @@ Status KafkaDataConsumerGroup::start_all(StreamLoadContext* ctx) {
// failed to append this msg, we must stop
LOG(WARNING) << "failed to append msg to pipe. grp: " << _grp_id;
eos = true;
{
std::unique_lock<std::mutex> lock(_mutex);
if (result_st.ok()) {
result_st = st;
}
}
}
delete msg;
} else {
Expand Down