You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2024/03/12 10:50:46 UTC
(doris) 03/07: [opt](routine-load) end Kafka consume when meets partition EOF #32046
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 6610307eb009c5905f4b906a017011504b58c0e4
Author: HHoflittlefish777 <77...@users.noreply.github.com>
AuthorDate: Tue Mar 12 16:05:28 2024 +0800
[opt](routine-load) end Kafka consume when meets partition EOF #32046
---
be/src/runtime/routine_load/data_consumer.cpp | 15 ++++++++++++++-
be/src/runtime/routine_load/data_consumer.h | 1 +
2 files changed, 15 insertions(+), 1 deletion(-)
diff --git a/be/src/runtime/routine_load/data_consumer.cpp b/be/src/runtime/routine_load/data_consumer.cpp
index 3706e31fb49..11c15c494fa 100644
--- a/be/src/runtime/routine_load/data_consumer.cpp
+++ b/be/src/runtime/routine_load/data_consumer.cpp
@@ -80,7 +80,7 @@ Status KafkaDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
};
RETURN_IF_ERROR(set_conf("metadata.broker.list", ctx->kafka_info->brokers));
- RETURN_IF_ERROR(set_conf("enable.partition.eof", "false"));
+ RETURN_IF_ERROR(set_conf("enable.partition.eof", "true"));
RETURN_IF_ERROR(set_conf("enable.auto.offset.store", "false"));
// TODO: set it larger than 0 after we set rd_kafka_conf_set_stats_cb()
RETURN_IF_ERROR(set_conf("statistics.interval.ms", "0"));
@@ -162,6 +162,7 @@ Status KafkaDataConsumer::assign_topic_partitions(
RdKafka::TopicPartition* tp1 =
RdKafka::TopicPartition::create(topic, entry.first, entry.second);
topic_partitions.push_back(tp1);
+ _consuming_partition_ids.insert(entry.first);
ss << "[" << entry.first << ": " << entry.second << "] ";
}
@@ -219,6 +220,9 @@ Status KafkaDataConsumer::group_consume(BlockingQueue<RdKafka::Message*>* queue,
consumer_watch.stop();
switch (msg->err()) {
case RdKafka::ERR_NO_ERROR:
+ if (_consuming_partition_ids.count(msg->partition()) <= 0) {
+ _consuming_partition_ids.insert(msg->partition());
+ }
if (msg->len() == 0) {
// ignore msg with length 0.
// put empty msg into queue will cause the load process shutting down.
@@ -245,6 +249,15 @@ Status KafkaDataConsumer::group_consume(BlockingQueue<RdKafka::Message*>* queue,
break;
}
[[fallthrough]];
+ case RdKafka::ERR__PARTITION_EOF: {
+ LOG(INFO) << "consumer meet partition eof: " << _id
+ << " partition offset: " << msg->offset();
+ _consuming_partition_ids.erase(msg->partition());
+ if (_consuming_partition_ids.size() <= 0) {
+ done = true;
+ }
+ break;
+ }
default:
LOG(WARNING) << "kafka consume failed: " << _id << ", msg: " << msg->errstr();
done = true;
diff --git a/be/src/runtime/routine_load/data_consumer.h b/be/src/runtime/routine_load/data_consumer.h
index 2a3f65f1993..0e1377681f4 100644
--- a/be/src/runtime/routine_load/data_consumer.h
+++ b/be/src/runtime/routine_load/data_consumer.h
@@ -160,6 +160,7 @@ private:
std::string _brokers;
std::string _topic;
std::unordered_map<std::string, std::string> _custom_properties;
+ std::set<int32_t> _consuming_partition_ids;
KafkaEventCb _k_event_cb;
RdKafka::KafkaConsumer* _k_consumer = nullptr;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org