You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "zshben (via GitHub)" <gi...@apache.org> on 2023/11/30 06:28:42 UTC
[PR] Pulsar test 1126 [doris]
zshben opened a new pull request, #27802:
URL: https://github.com/apache/doris/pull/27802
适配json拆分与数据入库
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Pulsar test 1126 [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #27802:
URL: https://github.com/apache/doris/pull/27802#discussion_r1410215411
##########
be/src/runtime/routine_load/data_consumer.cpp:
##########
@@ -452,4 +452,360 @@ bool KafkaDataConsumer::match(std::shared_ptr<StreamLoadContext> ctx) {
return true;
}
+// init pulsar consumer will only set common configs
+Status PulsarDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_init) {
+ // this consumer has already been initialized.
+ return Status::OK();
+ }
+
+ pulsar::ClientConfiguration config;
+ for (auto& item : ctx->pulsar_info->properties) {
+ if (item.first == "auth.token") {
+ config.setAuth(pulsar::AuthToken::createWithToken(item.second));
+ } else {
+ LOG(WARNING) << "Config " << item.first << " not supported for now.";
+ }
+
+ _custom_properties.emplace(item.first, item.second);
+ }
+
+ _p_client = new pulsar::Client(_service_url, config);
+
+ VLOG(3) << "finished to init pulsar consumer. " << ctx->brief();
+
+ _init = true;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::assign_partition(const std::string& partition, std::shared_ptr<StreamLoadContext> ctx,
+ int64_t initial_position) {
+ DCHECK(_p_client);
+
+ std::stringstream ss;
+ ss << "consumer: " << _id << ", grp: " << _grp_id << " topic: " << _topic
+ << ", subscription: " << _subscription << ", initial_position: " << initial_position;
+ LOG(INFO) << ss.str();
+
+ // do subscribe
+ pulsar::ConsumerConfiguration conf;
+ conf.setConsumerType(pulsar::ConsumerExclusive);
+
+ pulsar::Result result;
+ result = _p_client->subscribe(partition, _subscription, conf, _p_consumer);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to create pulsar consumer: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to create pulsar consumer: " +
+ std::string(pulsar::strResult(result)));
+ }
+
+ if (initial_position == InitialPosition::LATEST || initial_position == InitialPosition::EARLIEST) {
+ pulsar::InitialPosition p_initial_position = initial_position == InitialPosition::LATEST
+ ? pulsar::InitialPosition::InitialPositionLatest
+ : pulsar::InitialPosition::InitialPositionEarliest;
+ result = _p_consumer.seek(p_initial_position);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to reset the subscription: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to reset the subscription: " +
+ std::string(pulsar::strResult(result)));
+ }
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::group_consume(BlockingQueue<pulsar::Message*>* queue, int64_t max_running_time_ms) {
+ _last_visit_time = time(nullptr);
+ int64_t left_time = max_running_time_ms;
+ LOG(INFO) << "start pulsar consumer: " << _id << ", grp: " << _grp_id << ", max running time(ms): " << left_time;
+
+ int64_t received_rows = 0;
+ int64_t put_rows = 0;
+ Status st = Status::OK();
+ pulsar::MessageId msg_id;
+ std::string message_str;
+ MonotonicStopWatch consumer_watch;
+ MonotonicStopWatch watch;
+ watch.start();
+ while (true) {
+ {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_cancelled) {
+ break;
+ }
+ }
+
+ if (left_time <= 0) {
+ break;
+ }
+
+ bool done = false;
+ auto msg = std::make_unique<pulsar::Message>();
+ std::vector<const char*> rows;
+ std::string filter_data;
+ pulsar::Message* new_msg;
+ // consume 1 message at a time
+ consumer_watch.start();
+ pulsar::Result res = _p_consumer.receive(*(msg.get()), 30000 /* timeout, ms */);
+ consumer_watch.stop();
+ switch (res) {
+ case pulsar::ResultOk:
+ //filter invalid prefix of json
+ filter_data = substring_prefix_json(msg.get()->getDataAsString());
+ rows = convert_rows(filter_data.c_str());
+ msg_id = msg.get()->getMessageId();
+ message_str = msg.get()->getDataAsString();
+ if (received_rows == 0) {
+ LOG(INFO) << "receive first pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ }
+// if (msg.get()->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+// LOG(INFO) << "receive pulsar message "
+// << ", len: " << msg.get()->getLength()
+// << ", message id: " << msg_id
+// << ", pulsar consumer: " << _id
+// << ", grp: " << _grp_id
+// << ", rows size: " << rows.size();
+// }
+ for (const char* row : rows) {
+ pulsar::MessageBuilder messageBuilder;
+ size_t row_len = len_of_actual_data(row);
+ messageBuilder.setContent(row, row_len);
+ messageBuilder.setProperty("topicName",msg.get()->getTopicName());
+ new_msg = new pulsar::Message(messageBuilder.build());
+ new_msg->setMessageId(msg_id);
+
+ if (new_msg->getDataAsString().find("{\"") == std::string::npos) {
+ // ignore msg with length 0.
+ // put empty msg into queue will cause the load process shutting down.
+ LOG(INFO) << "pass error message: " << new_msg->getDataAsString();
+ break;
+ } else if (!queue->blocking_put(&(*new_msg))) {
+ // queue is shutdown
+ LOG(INFO) << "queue is shutdown, failed to blocking put" << new_msg->getDataAsString()
+ << ", len: " << new_msg->getLength()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ done = true;
+ } else {
+ ++put_rows;
+ }
+ ++received_rows;
+ }
+ delete msg.get();
+ msg.release(); // release the ownership, msg will be deleted after being processed
+ for (const char* row : rows) {
+ delete[] row;
+ }
+ rows.clear();
+ break;
+ case pulsar::ResultTimeout:
+ // leave the status as OK, because this may happened
+ // if there is no data in pulsar.
+ LOG(INFO) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume timeout.";
+ break;
+ default:
+ LOG(WARNING) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume failed: "
+ << ", errmsg: " << res;
+ done = true;
+ st = Status::InternalError(pulsar::strResult(res));
+ break;
+ }
+
+ left_time = max_running_time_ms - watch.elapsed_time() / 1000 / 1000;
Review Comment:
warning: 1000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
left_time = max_running_time_ms - watch.elapsed_time() / 1000 / 1000;
^
```
##########
be/src/runtime/routine_load/data_consumer.cpp:
##########
@@ -452,4 +452,360 @@
return true;
}
+// init pulsar consumer will only set common configs
+Status PulsarDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_init) {
+ // this consumer has already been initialized.
+ return Status::OK();
+ }
+
+ pulsar::ClientConfiguration config;
+ for (auto& item : ctx->pulsar_info->properties) {
+ if (item.first == "auth.token") {
+ config.setAuth(pulsar::AuthToken::createWithToken(item.second));
+ } else {
+ LOG(WARNING) << "Config " << item.first << " not supported for now.";
+ }
+
+ _custom_properties.emplace(item.first, item.second);
+ }
+
+ _p_client = new pulsar::Client(_service_url, config);
+
+ VLOG(3) << "finished to init pulsar consumer. " << ctx->brief();
+
+ _init = true;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::assign_partition(const std::string& partition, std::shared_ptr<StreamLoadContext> ctx,
+ int64_t initial_position) {
+ DCHECK(_p_client);
+
+ std::stringstream ss;
+ ss << "consumer: " << _id << ", grp: " << _grp_id << " topic: " << _topic
+ << ", subscription: " << _subscription << ", initial_position: " << initial_position;
+ LOG(INFO) << ss.str();
+
+ // do subscribe
+ pulsar::ConsumerConfiguration conf;
+ conf.setConsumerType(pulsar::ConsumerExclusive);
+
+ pulsar::Result result;
+ result = _p_client->subscribe(partition, _subscription, conf, _p_consumer);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to create pulsar consumer: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to create pulsar consumer: " +
+ std::string(pulsar::strResult(result)));
+ }
+
+ if (initial_position == InitialPosition::LATEST || initial_position == InitialPosition::EARLIEST) {
+ pulsar::InitialPosition p_initial_position = initial_position == InitialPosition::LATEST
+ ? pulsar::InitialPosition::InitialPositionLatest
+ : pulsar::InitialPosition::InitialPositionEarliest;
+ result = _p_consumer.seek(p_initial_position);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to reset the subscription: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to reset the subscription: " +
+ std::string(pulsar::strResult(result)));
+ }
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::group_consume(BlockingQueue<pulsar::Message*>* queue, int64_t max_running_time_ms) {
+ _last_visit_time = time(nullptr);
+ int64_t left_time = max_running_time_ms;
+ LOG(INFO) << "start pulsar consumer: " << _id << ", grp: " << _grp_id << ", max running time(ms): " << left_time;
+
+ int64_t received_rows = 0;
+ int64_t put_rows = 0;
+ Status st = Status::OK();
+ pulsar::MessageId msg_id;
+ std::string message_str;
+ MonotonicStopWatch consumer_watch;
+ MonotonicStopWatch watch;
+ watch.start();
+ while (true) {
+ {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_cancelled) {
+ break;
+ }
+ }
+
+ if (left_time <= 0) {
+ break;
+ }
+
+ bool done = false;
+ auto msg = std::make_unique<pulsar::Message>();
+ std::vector<const char*> rows;
+ std::string filter_data;
+ pulsar::Message* new_msg;
+ // consume 1 message at a time
+ consumer_watch.start();
+ pulsar::Result res = _p_consumer.receive(*(msg.get()), 30000 /* timeout, ms */);
+ consumer_watch.stop();
+ switch (res) {
+ case pulsar::ResultOk:
+ //filter invalid prefix of json
+ filter_data = substring_prefix_json(msg.get()->getDataAsString());
+ rows = convert_rows(filter_data.c_str());
+ msg_id = msg.get()->getMessageId();
+ message_str = msg.get()->getDataAsString();
+ if (received_rows == 0) {
+ LOG(INFO) << "receive first pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ }
+// if (msg.get()->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+// LOG(INFO) << "receive pulsar message "
+// << ", len: " << msg.get()->getLength()
+// << ", message id: " << msg_id
+// << ", pulsar consumer: " << _id
+// << ", grp: " << _grp_id
+// << ", rows size: " << rows.size();
+// }
+ for (const char* row : rows) {
+ pulsar::MessageBuilder messageBuilder;
+ size_t row_len = len_of_actual_data(row);
+ messageBuilder.setContent(row, row_len);
+ messageBuilder.setProperty("topicName",msg.get()->getTopicName());
+ new_msg = new pulsar::Message(messageBuilder.build());
+ new_msg->setMessageId(msg_id);
+
+ if (new_msg->getDataAsString().find("{\"") == std::string::npos) {
+ // ignore msg with length 0.
+ // put empty msg into queue will cause the load process shutting down.
+ LOG(INFO) << "pass error message: " << new_msg->getDataAsString();
+ break;
+ } else if (!queue->blocking_put(&(*new_msg))) {
+ // queue is shutdown
+ LOG(INFO) << "queue is shutdown, failed to blocking put" << new_msg->getDataAsString()
+ << ", len: " << new_msg->getLength()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ done = true;
+ } else {
+ ++put_rows;
+ }
+ ++received_rows;
+ }
+ delete msg.get();
+ msg.release(); // release the ownership, msg will be deleted after being processed
+ for (const char* row : rows) {
+ delete[] row;
+ }
+ rows.clear();
+ break;
+ case pulsar::ResultTimeout:
+ // leave the status as OK, because this may happened
+ // if there is no data in pulsar.
+ LOG(INFO) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume timeout.";
+ break;
+ default:
+ LOG(WARNING) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume failed: "
+ << ", errmsg: " << res;
+ done = true;
+ st = Status::InternalError(pulsar::strResult(res));
+ break;
+ }
+
+ left_time = max_running_time_ms - watch.elapsed_time() / 1000 / 1000;
+ if (done) {
+ break;
+ }
+ }
+
+ LOG(INFO) << "receive last pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id;
+
+// LOG(INFO) << "start do ack of msg_id :" << msg_id;
+// pulsar::Result ack = _p_consumer.acknowledgeCumulative(msg_id);
+// if (ack != pulsar::ResultOk) {
+// LOG(WARNING) << "failed do ack of msg_id :" << msg_id << ", consumer : " << _id;
+// } else {
+// LOG(INFO) << "finish do ack of msg_id :" << msg_id << ", consumer : " << _id;
+// }
+
+ LOG(INFO) << "pulsar consume done: " << _id << ", grp: " << _grp_id << ". cancelled: " << _cancelled
+ << ", left time(ms): " << left_time << ", total cost(ms): " << watch.elapsed_time() / 1000 / 1000
+ << ", consume cost(ms): " << consumer_watch.elapsed_time() / 1000 / 1000
+ << ", received rows: " << received_rows << ", put rows: " << put_rows
+ << ", last message id: " << msg_id;
+
+ return st;
+}
+
+const std::string& PulsarDataConsumer::get_partition() {
+ _last_visit_time = time(nullptr);
+ return _p_consumer.getTopic();
+}
+
+Status PulsarDataConsumer::get_partition_backlog(int64_t* backlog) {
+ _last_visit_time = time(nullptr);
+ pulsar::BrokerConsumerStats broker_consumer_stats;
+ pulsar::Result result = _p_consumer.getBrokerConsumerStats(broker_consumer_stats);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "Failed to get broker consumer stats: "
+ << ", err: " << result;
+ return Status::InternalError("Failed to get broker consumer stats: " + std::string(pulsar::strResult(result)));
+ }
+ *backlog = broker_consumer_stats.getMsgBacklog();
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::get_topic_partition(std::vector<std::string>* partitions) {
+ _last_visit_time = time(nullptr);
+ pulsar::Result result = _p_client->getPartitionsForTopic(_topic, *partitions);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "Failed to get partitions for topic: " << _topic << ", err: " << result;
+ return Status::InternalError("Failed to get partitions for topic: " + std::string(pulsar::strResult(result)));
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::cancel(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (!_init) {
+ return Status::InternalError("consumer is not initialized");
+ }
+
+ _cancelled = true;
+ LOG(INFO) << "pulsar consumer cancelled. " << _id;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::reset() {
+ std::unique_lock<std::mutex> l(_lock);
+ _cancelled = false;
+ _p_consumer.close();
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::acknowledge_cumulative(pulsar::MessageId& message_id) {
Review Comment:
warning: method 'acknowledge_cumulative' can be made static [readability-convert-member-functions-to-static]
```suggestion
static Status PulsarDataConsumer::acknowledge_cumulative(pulsar::MessageId& message_id) {
```
##########
be/src/runtime/routine_load/data_consumer.cpp:
##########
@@ -452,4 +452,360 @@
return true;
}
+// init pulsar consumer will only set common configs
+Status PulsarDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_init) {
+ // this consumer has already been initialized.
+ return Status::OK();
+ }
+
+ pulsar::ClientConfiguration config;
+ for (auto& item : ctx->pulsar_info->properties) {
+ if (item.first == "auth.token") {
+ config.setAuth(pulsar::AuthToken::createWithToken(item.second));
+ } else {
+ LOG(WARNING) << "Config " << item.first << " not supported for now.";
+ }
+
+ _custom_properties.emplace(item.first, item.second);
+ }
+
+ _p_client = new pulsar::Client(_service_url, config);
+
+ VLOG(3) << "finished to init pulsar consumer. " << ctx->brief();
+
+ _init = true;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::assign_partition(const std::string& partition, std::shared_ptr<StreamLoadContext> ctx,
+ int64_t initial_position) {
+ DCHECK(_p_client);
+
+ std::stringstream ss;
+ ss << "consumer: " << _id << ", grp: " << _grp_id << " topic: " << _topic
+ << ", subscription: " << _subscription << ", initial_position: " << initial_position;
+ LOG(INFO) << ss.str();
+
+ // do subscribe
+ pulsar::ConsumerConfiguration conf;
+ conf.setConsumerType(pulsar::ConsumerExclusive);
+
+ pulsar::Result result;
+ result = _p_client->subscribe(partition, _subscription, conf, _p_consumer);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to create pulsar consumer: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to create pulsar consumer: " +
+ std::string(pulsar::strResult(result)));
+ }
+
+ if (initial_position == InitialPosition::LATEST || initial_position == InitialPosition::EARLIEST) {
+ pulsar::InitialPosition p_initial_position = initial_position == InitialPosition::LATEST
+ ? pulsar::InitialPosition::InitialPositionLatest
+ : pulsar::InitialPosition::InitialPositionEarliest;
+ result = _p_consumer.seek(p_initial_position);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to reset the subscription: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to reset the subscription: " +
+ std::string(pulsar::strResult(result)));
+ }
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::group_consume(BlockingQueue<pulsar::Message*>* queue, int64_t max_running_time_ms) {
+ _last_visit_time = time(nullptr);
+ int64_t left_time = max_running_time_ms;
+ LOG(INFO) << "start pulsar consumer: " << _id << ", grp: " << _grp_id << ", max running time(ms): " << left_time;
+
+ int64_t received_rows = 0;
+ int64_t put_rows = 0;
+ Status st = Status::OK();
+ pulsar::MessageId msg_id;
+ std::string message_str;
+ MonotonicStopWatch consumer_watch;
+ MonotonicStopWatch watch;
+ watch.start();
+ while (true) {
+ {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_cancelled) {
+ break;
+ }
+ }
+
+ if (left_time <= 0) {
+ break;
+ }
+
+ bool done = false;
+ auto msg = std::make_unique<pulsar::Message>();
+ std::vector<const char*> rows;
+ std::string filter_data;
+ pulsar::Message* new_msg;
+ // consume 1 message at a time
+ consumer_watch.start();
+ pulsar::Result res = _p_consumer.receive(*(msg.get()), 30000 /* timeout, ms */);
+ consumer_watch.stop();
+ switch (res) {
+ case pulsar::ResultOk:
+ //filter invalid prefix of json
+ filter_data = substring_prefix_json(msg.get()->getDataAsString());
+ rows = convert_rows(filter_data.c_str());
+ msg_id = msg.get()->getMessageId();
+ message_str = msg.get()->getDataAsString();
+ if (received_rows == 0) {
+ LOG(INFO) << "receive first pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ }
+// if (msg.get()->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+// LOG(INFO) << "receive pulsar message "
+// << ", len: " << msg.get()->getLength()
+// << ", message id: " << msg_id
+// << ", pulsar consumer: " << _id
+// << ", grp: " << _grp_id
+// << ", rows size: " << rows.size();
+// }
+ for (const char* row : rows) {
+ pulsar::MessageBuilder messageBuilder;
+ size_t row_len = len_of_actual_data(row);
+ messageBuilder.setContent(row, row_len);
+ messageBuilder.setProperty("topicName",msg.get()->getTopicName());
+ new_msg = new pulsar::Message(messageBuilder.build());
+ new_msg->setMessageId(msg_id);
+
+ if (new_msg->getDataAsString().find("{\"") == std::string::npos) {
+ // ignore msg with length 0.
+ // put empty msg into queue will cause the load process shutting down.
+ LOG(INFO) << "pass error message: " << new_msg->getDataAsString();
+ break;
+ } else if (!queue->blocking_put(&(*new_msg))) {
+ // queue is shutdown
+ LOG(INFO) << "queue is shutdown, failed to blocking put" << new_msg->getDataAsString()
+ << ", len: " << new_msg->getLength()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ done = true;
+ } else {
+ ++put_rows;
+ }
+ ++received_rows;
+ }
+ delete msg.get();
+ msg.release(); // release the ownership, msg will be deleted after being processed
+ for (const char* row : rows) {
+ delete[] row;
+ }
+ rows.clear();
+ break;
+ case pulsar::ResultTimeout:
+ // leave the status as OK, because this may happened
+ // if there is no data in pulsar.
+ LOG(INFO) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume timeout.";
+ break;
+ default:
+ LOG(WARNING) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume failed: "
+ << ", errmsg: " << res;
+ done = true;
+ st = Status::InternalError(pulsar::strResult(res));
+ break;
+ }
+
+ left_time = max_running_time_ms - watch.elapsed_time() / 1000 / 1000;
Review Comment:
warning: 1000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
left_time = max_running_time_ms - watch.elapsed_time() / 1000 / 1000;
^
```
##########
be/src/runtime/routine_load/data_consumer.cpp:
##########
@@ -452,4 +452,360 @@
return true;
}
+// init pulsar consumer will only set common configs
+Status PulsarDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_init) {
+ // this consumer has already been initialized.
+ return Status::OK();
+ }
+
+ pulsar::ClientConfiguration config;
+ for (auto& item : ctx->pulsar_info->properties) {
+ if (item.first == "auth.token") {
+ config.setAuth(pulsar::AuthToken::createWithToken(item.second));
+ } else {
+ LOG(WARNING) << "Config " << item.first << " not supported for now.";
+ }
+
+ _custom_properties.emplace(item.first, item.second);
+ }
+
+ _p_client = new pulsar::Client(_service_url, config);
+
+ VLOG(3) << "finished to init pulsar consumer. " << ctx->brief();
+
+ _init = true;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::assign_partition(const std::string& partition, std::shared_ptr<StreamLoadContext> ctx,
+ int64_t initial_position) {
+ DCHECK(_p_client);
+
+ std::stringstream ss;
+ ss << "consumer: " << _id << ", grp: " << _grp_id << " topic: " << _topic
+ << ", subscription: " << _subscription << ", initial_position: " << initial_position;
+ LOG(INFO) << ss.str();
+
+ // do subscribe
+ pulsar::ConsumerConfiguration conf;
+ conf.setConsumerType(pulsar::ConsumerExclusive);
+
+ pulsar::Result result;
+ result = _p_client->subscribe(partition, _subscription, conf, _p_consumer);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to create pulsar consumer: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to create pulsar consumer: " +
+ std::string(pulsar::strResult(result)));
+ }
+
+ if (initial_position == InitialPosition::LATEST || initial_position == InitialPosition::EARLIEST) {
+ pulsar::InitialPosition p_initial_position = initial_position == InitialPosition::LATEST
+ ? pulsar::InitialPosition::InitialPositionLatest
+ : pulsar::InitialPosition::InitialPositionEarliest;
+ result = _p_consumer.seek(p_initial_position);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to reset the subscription: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to reset the subscription: " +
+ std::string(pulsar::strResult(result)));
+ }
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::group_consume(BlockingQueue<pulsar::Message*>* queue, int64_t max_running_time_ms) {
+ _last_visit_time = time(nullptr);
+ int64_t left_time = max_running_time_ms;
+ LOG(INFO) << "start pulsar consumer: " << _id << ", grp: " << _grp_id << ", max running time(ms): " << left_time;
+
+ int64_t received_rows = 0;
+ int64_t put_rows = 0;
+ Status st = Status::OK();
+ pulsar::MessageId msg_id;
+ std::string message_str;
+ MonotonicStopWatch consumer_watch;
+ MonotonicStopWatch watch;
+ watch.start();
+ while (true) {
+ {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_cancelled) {
+ break;
+ }
+ }
+
+ if (left_time <= 0) {
+ break;
+ }
+
+ bool done = false;
+ auto msg = std::make_unique<pulsar::Message>();
+ std::vector<const char*> rows;
+ std::string filter_data;
+ pulsar::Message* new_msg;
+ // consume 1 message at a time
+ consumer_watch.start();
+ pulsar::Result res = _p_consumer.receive(*(msg.get()), 30000 /* timeout, ms */);
+ consumer_watch.stop();
+ switch (res) {
+ case pulsar::ResultOk:
+ //filter invalid prefix of json
+ filter_data = substring_prefix_json(msg.get()->getDataAsString());
+ rows = convert_rows(filter_data.c_str());
+ msg_id = msg.get()->getMessageId();
+ message_str = msg.get()->getDataAsString();
+ if (received_rows == 0) {
+ LOG(INFO) << "receive first pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ }
+// if (msg.get()->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+// LOG(INFO) << "receive pulsar message "
+// << ", len: " << msg.get()->getLength()
+// << ", message id: " << msg_id
+// << ", pulsar consumer: " << _id
+// << ", grp: " << _grp_id
+// << ", rows size: " << rows.size();
+// }
+ for (const char* row : rows) {
+ pulsar::MessageBuilder messageBuilder;
+ size_t row_len = len_of_actual_data(row);
+ messageBuilder.setContent(row, row_len);
+ messageBuilder.setProperty("topicName",msg.get()->getTopicName());
+ new_msg = new pulsar::Message(messageBuilder.build());
+ new_msg->setMessageId(msg_id);
+
+ if (new_msg->getDataAsString().find("{\"") == std::string::npos) {
+ // ignore msg with length 0.
+ // put empty msg into queue will cause the load process shutting down.
+ LOG(INFO) << "pass error message: " << new_msg->getDataAsString();
+ break;
+ } else if (!queue->blocking_put(&(*new_msg))) {
+ // queue is shutdown
+ LOG(INFO) << "queue is shutdown, failed to blocking put" << new_msg->getDataAsString()
+ << ", len: " << new_msg->getLength()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ done = true;
+ } else {
+ ++put_rows;
+ }
+ ++received_rows;
+ }
+ delete msg.get();
+ msg.release(); // release the ownership, msg will be deleted after being processed
+ for (const char* row : rows) {
+ delete[] row;
+ }
+ rows.clear();
+ break;
+ case pulsar::ResultTimeout:
+ // leave the status as OK, because this may happened
+ // if there is no data in pulsar.
+ LOG(INFO) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume timeout.";
+ break;
+ default:
+ LOG(WARNING) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume failed: "
+ << ", errmsg: " << res;
+ done = true;
+ st = Status::InternalError(pulsar::strResult(res));
+ break;
+ }
+
+ left_time = max_running_time_ms - watch.elapsed_time() / 1000 / 1000;
+ if (done) {
+ break;
+ }
+ }
+
+ LOG(INFO) << "receive last pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id;
+
+// LOG(INFO) << "start do ack of msg_id :" << msg_id;
+// pulsar::Result ack = _p_consumer.acknowledgeCumulative(msg_id);
+// if (ack != pulsar::ResultOk) {
+// LOG(WARNING) << "failed do ack of msg_id :" << msg_id << ", consumer : " << _id;
+// } else {
+// LOG(INFO) << "finish do ack of msg_id :" << msg_id << ", consumer : " << _id;
+// }
+
+ LOG(INFO) << "pulsar consume done: " << _id << ", grp: " << _grp_id << ". cancelled: " << _cancelled
+ << ", left time(ms): " << left_time << ", total cost(ms): " << watch.elapsed_time() / 1000 / 1000
+ << ", consume cost(ms): " << consumer_watch.elapsed_time() / 1000 / 1000
+ << ", received rows: " << received_rows << ", put rows: " << put_rows
+ << ", last message id: " << msg_id;
+
+ return st;
+}
+
+const std::string& PulsarDataConsumer::get_partition() {
+ _last_visit_time = time(nullptr);
+ return _p_consumer.getTopic();
+}
+
+Status PulsarDataConsumer::get_partition_backlog(int64_t* backlog) {
+ _last_visit_time = time(nullptr);
+ pulsar::BrokerConsumerStats broker_consumer_stats;
+ pulsar::Result result = _p_consumer.getBrokerConsumerStats(broker_consumer_stats);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "Failed to get broker consumer stats: "
+ << ", err: " << result;
+ return Status::InternalError("Failed to get broker consumer stats: " + std::string(pulsar::strResult(result)));
+ }
+ *backlog = broker_consumer_stats.getMsgBacklog();
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::get_topic_partition(std::vector<std::string>* partitions) {
+ _last_visit_time = time(nullptr);
+ pulsar::Result result = _p_client->getPartitionsForTopic(_topic, *partitions);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "Failed to get partitions for topic: " << _topic << ", err: " << result;
+ return Status::InternalError("Failed to get partitions for topic: " + std::string(pulsar::strResult(result)));
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::cancel(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (!_init) {
+ return Status::InternalError("consumer is not initialized");
+ }
+
+ _cancelled = true;
+ LOG(INFO) << "pulsar consumer cancelled. " << _id;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::reset() {
+ std::unique_lock<std::mutex> l(_lock);
+ _cancelled = false;
+ _p_consumer.close();
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::acknowledge_cumulative(pulsar::MessageId& message_id) {
+ pulsar::Result res = _p_consumer.acknowledgeCumulative(message_id);
+ if (res != pulsar::ResultOk) {
+ std::stringstream ss;
+ ss << "failed to acknowledge pulsar message : " << res;
+ LOG(WARNING) << "failed to acknowledge pulsar message : " << res
+ << ",pulsar consumer :" << _id
+ << ",pulsar group :" << _grp_id;
+ return Status::InternalError(ss.str());
+ }
+ LOG(INFO) << "message_id :" << message_id
+ << ",pulsar consumer :" << _id
+ << ",pulsar group :" << _grp_id;
+ return Status::OK();
+}
+
+bool PulsarDataConsumer::match(std::shared_ptr<StreamLoadContext> ctx) {
+ if (ctx->load_src_type != TLoadSourceType::PULSAR) {
+ return false;
+ }
+ if (_service_url != ctx->pulsar_info->service_url || _topic != ctx->pulsar_info->topic ||
+ _subscription != ctx->pulsar_info->subscription) {
+ return false;
+ }
+ // check properties
+ if (_custom_properties.size() != ctx->pulsar_info->properties.size()) {
+ return false;
+ }
+ for (auto& item : ctx->pulsar_info->properties) {
+ auto it = _custom_properties.find(item.first);
+ if (it == _custom_properties.end() || it->second != item.second) {
+ return false;
+ }
+ }
+
+ return true;
+}
+
+std::string PulsarDataConsumer::substring_prefix_json(std::string data) {
+ // 找到以 "{" 开头的位置
+ size_t startPos = data.find("{\"");
+
+ // 如果找到了，则截取并返回子字符串
+ if (startPos != std::string::npos) {
+ return data.substr(startPos);
+ } else {
+ return data;
+ }
+}
+
+size_t PulsarDataConsumer::len_of_actual_data(const char* data) {
+ size_t length = 0;
Review Comment:
warning: method 'len_of_actual_data' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/routine_load/data_consumer.h:212:
```diff
- size_t len_of_actual_data(const char* data);
+ static size_t len_of_actual_data(const char* data);
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
+ int64_t received_rows = 0;
+ int64_t left_bytes = ctx->max_batch_size;
+
+ std::shared_ptr<io::PulsarConsumerPipe> pulsar_pipe = std::static_pointer_cast<io::PulsarConsumerPipe>(ctx->body_sink);
+
+ LOG(INFO) << "start consumer group: " << _grp_id << ". max time(ms): " << left_time
+ << ", batch size: " << left_bytes << ". " << ctx->brief();
+
+ // copy one
+ std::map<std::string, pulsar::MessageId> ack_offset = ctx->pulsar_info->ack_offset;
+
+ //improve performance
+ Status (io::PulsarConsumerPipe::*append_data)(const char* data, size_t size);
+ if (ctx->format == TFileFormatType::FORMAT_JSON) {
+ append_data = &io::PulsarConsumerPipe::append_json;
+ } else {
+ append_data = &io::PulsarConsumerPipe::append_with_line_delimiter;
+ }
+
+ MonotonicStopWatch watch;
+ watch.start();
+ bool eos = false;
+ while (true) {
+ if (eos || left_time <= 0 || left_bytes <= 0) {
+ LOG(INFO) << "consumer group done: " << _grp_id
+ << ". consume time(ms)=" << ctx->max_interval_s * 1000 - left_time
+ << ", received rows=" << received_rows << ", received bytes=" << ctx->max_batch_size - left_bytes
+ << ", eos: " << eos << ", left_time: " << left_time << ", left_bytes: " << left_bytes
+ << ", blocking get time(us): " << _queue.total_get_wait_time() / 1000
+ << ", blocking put time(us): " << _queue.total_put_wait_time() / 1000;
+
+ // cancel all consumers
+ for (auto& consumer : _consumers) {
+ static_cast<void>(consumer->cancel(ctx));
+ }
+
+ // sleep 3s for waitting consumer cancel done
+ std::this_thread::sleep_for(std::chrono::seconds(3));
+ // shutdown queue
+ _queue.shutdown();
+
+ // waiting all threads finished
+ _thread_pool.shutdown();
+ _thread_pool.join();
+
+ if (!result_st.ok()) {
+ // some consumers encounter errors, cancel this task
+ LOG(WARNING) << "Failed. Cancelled append msg to pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel(result_st.to_string());
+ return result_st;
+ }
+
+ if (left_bytes == ctx->max_batch_size) {
+ // nothing to be consumed, we have to cancel it, because
+ // we do not allow finishing stream load pipe without data
+ LOG(WARNING) << "Cancelled append msg to pulsar pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel("Cancelled");
+ return Status::InternalError("Cancelled");
+ } else {
+ DCHECK(left_bytes < ctx->max_batch_size);
+ static_cast<void>(pulsar_pipe->finish());
+ ctx->pulsar_info->ack_offset = std::move(ack_offset);
+ ctx->receive_bytes = ctx->max_batch_size - left_bytes;
+ get_backlog_nums(ctx);
+ acknowledge_cumulative(ctx);
+ return Status::OK();
+ }
+ }
+
+ pulsar::Message* msg;
+ bool res = _queue.blocking_get(&msg);
+ if (res) {
+ std::string partition = msg->getProperty("topicName");
+ pulsar::MessageId msg_id = msg->getMessageId();
+ std::size_t len = msg->getLength();
+
+ VLOG(3) << "get pulsar message:" << msg->getDataAsString()
+ << ", partition: " << partition << ", message id: " << msg_id
+ << ", len: " << len << ", size: " << msg->getDataAsString().size();
+
+ Status st = (pulsar_pipe.get()->*append_data)(msg->getDataAsString().c_str(),
+ msg->getDataAsString().size());
+ if (st.ok()) {
+ received_rows++;
+ // len of receive origin message from pulsar
+ left_bytes -= len;
+ if (ack_offset.find(partition) != ack_offset.end()) {
+ if (msg->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
Review Comment:
warning: escaped string literal can be written as a raw string literal [modernize-raw-string-literal]
```suggestion
if (msg->getDataAsString().find(R"("country":"PL")") != std::string::npos) {
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
+ int64_t received_rows = 0;
+ int64_t left_bytes = ctx->max_batch_size;
+
+ std::shared_ptr<io::PulsarConsumerPipe> pulsar_pipe = std::static_pointer_cast<io::PulsarConsumerPipe>(ctx->body_sink);
+
+ LOG(INFO) << "start consumer group: " << _grp_id << ". max time(ms): " << left_time
+ << ", batch size: " << left_bytes << ". " << ctx->brief();
+
+ // copy one
+ std::map<std::string, pulsar::MessageId> ack_offset = ctx->pulsar_info->ack_offset;
+
+ //improve performance
+ Status (io::PulsarConsumerPipe::*append_data)(const char* data, size_t size);
+ if (ctx->format == TFileFormatType::FORMAT_JSON) {
+ append_data = &io::PulsarConsumerPipe::append_json;
+ } else {
+ append_data = &io::PulsarConsumerPipe::append_with_line_delimiter;
+ }
+
+ MonotonicStopWatch watch;
+ watch.start();
+ bool eos = false;
+ while (true) {
+ if (eos || left_time <= 0 || left_bytes <= 0) {
+ LOG(INFO) << "consumer group done: " << _grp_id
+ << ". consume time(ms)=" << ctx->max_interval_s * 1000 - left_time
+ << ", received rows=" << received_rows << ", received bytes=" << ctx->max_batch_size - left_bytes
+ << ", eos: " << eos << ", left_time: " << left_time << ", left_bytes: " << left_bytes
+ << ", blocking get time(us): " << _queue.total_get_wait_time() / 1000
+ << ", blocking put time(us): " << _queue.total_put_wait_time() / 1000;
+
+ // cancel all consumers
+ for (auto& consumer : _consumers) {
+ static_cast<void>(consumer->cancel(ctx));
+ }
+
+ // sleep 3s for waitting consumer cancel done
+ std::this_thread::sleep_for(std::chrono::seconds(3));
+ // shutdown queue
+ _queue.shutdown();
+
+ // waiting all threads finished
+ _thread_pool.shutdown();
+ _thread_pool.join();
+
+ if (!result_st.ok()) {
+ // some consumers encounter errors, cancel this task
+ LOG(WARNING) << "Failed. Cancelled append msg to pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel(result_st.to_string());
+ return result_st;
+ }
+
+ if (left_bytes == ctx->max_batch_size) {
+ // nothing to be consumed, we have to cancel it, because
+ // we do not allow finishing stream load pipe without data
+ LOG(WARNING) << "Cancelled append msg to pulsar pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel("Cancelled");
+ return Status::InternalError("Cancelled");
+ } else {
+ DCHECK(left_bytes < ctx->max_batch_size);
+ static_cast<void>(pulsar_pipe->finish());
+ ctx->pulsar_info->ack_offset = std::move(ack_offset);
+ ctx->receive_bytes = ctx->max_batch_size - left_bytes;
+ get_backlog_nums(ctx);
+ acknowledge_cumulative(ctx);
+ return Status::OK();
+ }
Review Comment:
warning: do not use 'else' after 'return' [readability-else-after-return]
```suggestion
} DCHECK(left_bytes < ctx->max_batch_size);
static_cast<void>(pulsar_pipe->finish());
ctx->pulsar_info->ack_offset = std::move(ack_offset);
ctx->receive_bytes = ctx->max_batch_size - left_bytes;
get_backlog_nums(ctx);
acknowledge_cumulative(ctx);
return Status::OK();
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
+ int64_t received_rows = 0;
+ int64_t left_bytes = ctx->max_batch_size;
+
+ std::shared_ptr<io::PulsarConsumerPipe> pulsar_pipe = std::static_pointer_cast<io::PulsarConsumerPipe>(ctx->body_sink);
+
+ LOG(INFO) << "start consumer group: " << _grp_id << ". max time(ms): " << left_time
+ << ", batch size: " << left_bytes << ". " << ctx->brief();
+
+ // copy one
+ std::map<std::string, pulsar::MessageId> ack_offset = ctx->pulsar_info->ack_offset;
+
+ //improve performance
+ Status (io::PulsarConsumerPipe::*append_data)(const char* data, size_t size);
+ if (ctx->format == TFileFormatType::FORMAT_JSON) {
+ append_data = &io::PulsarConsumerPipe::append_json;
+ } else {
+ append_data = &io::PulsarConsumerPipe::append_with_line_delimiter;
+ }
+
+ MonotonicStopWatch watch;
+ watch.start();
+ bool eos = false;
+ while (true) {
+ if (eos || left_time <= 0 || left_bytes <= 0) {
+ LOG(INFO) << "consumer group done: " << _grp_id
+ << ". consume time(ms)=" << ctx->max_interval_s * 1000 - left_time
+ << ", received rows=" << received_rows << ", received bytes=" << ctx->max_batch_size - left_bytes
+ << ", eos: " << eos << ", left_time: " << left_time << ", left_bytes: " << left_bytes
+ << ", blocking get time(us): " << _queue.total_get_wait_time() / 1000
+ << ", blocking put time(us): " << _queue.total_put_wait_time() / 1000;
+
+ // cancel all consumers
+ for (auto& consumer : _consumers) {
+ static_cast<void>(consumer->cancel(ctx));
+ }
+
+ // sleep 3s for waitting consumer cancel done
+ std::this_thread::sleep_for(std::chrono::seconds(3));
+ // shutdown queue
+ _queue.shutdown();
+
+ // waiting all threads finished
+ _thread_pool.shutdown();
+ _thread_pool.join();
+
+ if (!result_st.ok()) {
+ // some consumers encounter errors, cancel this task
+ LOG(WARNING) << "Failed. Cancelled append msg to pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel(result_st.to_string());
+ return result_st;
+ }
+
+ if (left_bytes == ctx->max_batch_size) {
+ // nothing to be consumed, we have to cancel it, because
+ // we do not allow finishing stream load pipe without data
+ LOG(WARNING) << "Cancelled append msg to pulsar pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel("Cancelled");
+ return Status::InternalError("Cancelled");
+ } else {
+ DCHECK(left_bytes < ctx->max_batch_size);
+ static_cast<void>(pulsar_pipe->finish());
+ ctx->pulsar_info->ack_offset = std::move(ack_offset);
+ ctx->receive_bytes = ctx->max_batch_size - left_bytes;
+ get_backlog_nums(ctx);
+ acknowledge_cumulative(ctx);
+ return Status::OK();
+ }
+ }
+
+ pulsar::Message* msg;
+ bool res = _queue.blocking_get(&msg);
+ if (res) {
+ std::string partition = msg->getProperty("topicName");
+ pulsar::MessageId msg_id = msg->getMessageId();
+ std::size_t len = msg->getLength();
+
+ VLOG(3) << "get pulsar message:" << msg->getDataAsString()
+ << ", partition: " << partition << ", message id: " << msg_id
+ << ", len: " << len << ", size: " << msg->getDataAsString().size();
+
+ Status st = (pulsar_pipe.get()->*append_data)(msg->getDataAsString().c_str(),
+ msg->getDataAsString().size());
+ if (st.ok()) {
+ received_rows++;
+ // len of receive origin message from pulsar
+ left_bytes -= len;
+ if (ack_offset.find(partition) != ack_offset.end()) {
+ if (msg->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+ LOG(INFO) << "old message id: " << ack_offset[partition]
+ << ", new message id: " << msg_id;
+ }
+ if (ack_offset[partition] < msg_id) {
+ ack_offset[partition] = msg_id;
+ }
+ } else {
+ ack_offset[partition] = msg_id;
+ }
+ VLOG(3) << "consume partition" << partition << " - " << msg_id;
+ } else {
+ // failed to append this msg, we must stop
+ LOG(WARNING) << "failed to append msg to pipe. grp: " << _grp_id << ", errmsg=" << st.to_string();
+ eos = true;
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ if (result_st.ok()) {
+ result_st = st;
+ }
+ }
+ }
+ delete msg;
+ } else {
+ // queue is empty and shutdown
+ eos = true;
+ }
+
+ left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
+ }
+
+ return Status::OK();
+}
+
+void PulsarDataConsumerGroup::actual_consume(const std::shared_ptr<DataConsumer>& consumer,
Review Comment:
warning: method 'actual_consume' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void PulsarDataConsumerGroup::actual_consume(const std::shared_ptr<DataConsumer>& consumer,
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
+ int64_t received_rows = 0;
+ int64_t left_bytes = ctx->max_batch_size;
+
+ std::shared_ptr<io::PulsarConsumerPipe> pulsar_pipe = std::static_pointer_cast<io::PulsarConsumerPipe>(ctx->body_sink);
+
+ LOG(INFO) << "start consumer group: " << _grp_id << ". max time(ms): " << left_time
+ << ", batch size: " << left_bytes << ". " << ctx->brief();
+
+ // copy one
+ std::map<std::string, pulsar::MessageId> ack_offset = ctx->pulsar_info->ack_offset;
+
+ //improve performance
+ Status (io::PulsarConsumerPipe::*append_data)(const char* data, size_t size);
+ if (ctx->format == TFileFormatType::FORMAT_JSON) {
+ append_data = &io::PulsarConsumerPipe::append_json;
+ } else {
+ append_data = &io::PulsarConsumerPipe::append_with_line_delimiter;
+ }
+
+ MonotonicStopWatch watch;
+ watch.start();
+ bool eos = false;
+ while (true) {
+ if (eos || left_time <= 0 || left_bytes <= 0) {
+ LOG(INFO) << "consumer group done: " << _grp_id
+ << ". consume time(ms)=" << ctx->max_interval_s * 1000 - left_time
+ << ", received rows=" << received_rows << ", received bytes=" << ctx->max_batch_size - left_bytes
+ << ", eos: " << eos << ", left_time: " << left_time << ", left_bytes: " << left_bytes
+ << ", blocking get time(us): " << _queue.total_get_wait_time() / 1000
+ << ", blocking put time(us): " << _queue.total_put_wait_time() / 1000;
+
+ // cancel all consumers
+ for (auto& consumer : _consumers) {
+ static_cast<void>(consumer->cancel(ctx));
+ }
+
+ // sleep 3s for waitting consumer cancel done
+ std::this_thread::sleep_for(std::chrono::seconds(3));
+ // shutdown queue
+ _queue.shutdown();
+
+ // waiting all threads finished
+ _thread_pool.shutdown();
+ _thread_pool.join();
+
+ if (!result_st.ok()) {
+ // some consumers encounter errors, cancel this task
+ LOG(WARNING) << "Failed. Cancelled append msg to pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel(result_st.to_string());
+ return result_st;
+ }
+
+ if (left_bytes == ctx->max_batch_size) {
+ // nothing to be consumed, we have to cancel it, because
+ // we do not allow finishing stream load pipe without data
+ LOG(WARNING) << "Cancelled append msg to pulsar pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel("Cancelled");
+ return Status::InternalError("Cancelled");
+ } else {
+ DCHECK(left_bytes < ctx->max_batch_size);
+ static_cast<void>(pulsar_pipe->finish());
+ ctx->pulsar_info->ack_offset = std::move(ack_offset);
+ ctx->receive_bytes = ctx->max_batch_size - left_bytes;
+ get_backlog_nums(ctx);
+ acknowledge_cumulative(ctx);
+ return Status::OK();
+ }
+ }
+
+ pulsar::Message* msg;
+ bool res = _queue.blocking_get(&msg);
+ if (res) {
+ std::string partition = msg->getProperty("topicName");
+ pulsar::MessageId msg_id = msg->getMessageId();
+ std::size_t len = msg->getLength();
+
+ VLOG(3) << "get pulsar message:" << msg->getDataAsString()
+ << ", partition: " << partition << ", message id: " << msg_id
+ << ", len: " << len << ", size: " << msg->getDataAsString().size();
+
+ Status st = (pulsar_pipe.get()->*append_data)(msg->getDataAsString().c_str(),
+ msg->getDataAsString().size());
+ if (st.ok()) {
+ received_rows++;
+ // len of receive origin message from pulsar
+ left_bytes -= len;
+ if (ack_offset.find(partition) != ack_offset.end()) {
+ if (msg->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+ LOG(INFO) << "old message id: " << ack_offset[partition]
+ << ", new message id: " << msg_id;
+ }
+ if (ack_offset[partition] < msg_id) {
+ ack_offset[partition] = msg_id;
+ }
+ } else {
+ ack_offset[partition] = msg_id;
+ }
+ VLOG(3) << "consume partition" << partition << " - " << msg_id;
+ } else {
+ // failed to append this msg, we must stop
+ LOG(WARNING) << "failed to append msg to pipe. grp: " << _grp_id << ", errmsg=" << st.to_string();
+ eos = true;
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ if (result_st.ok()) {
+ result_st = st;
+ }
+ }
+ }
+ delete msg;
+ } else {
+ // queue is empty and shutdown
+ eos = true;
+ }
+
+ left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
+ }
+
+ return Status::OK();
+}
+
+void PulsarDataConsumerGroup::actual_consume(const std::shared_ptr<DataConsumer>& consumer,
+ BlockingQueue<pulsar::Message*>* queue, int64_t max_running_time_ms,
+ const ConsumeFinishCallback& cb) {
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->group_consume(queue, max_running_time_ms);
+ cb(st);
+}
+
+void PulsarDataConsumerGroup::get_backlog_nums(std::shared_ptr<StreamLoadContext> ctx) {
+ for (auto& consumer : _consumers) {
+ // get backlog num
+ int64_t backlog_num;
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->get_partition_backlog(&backlog_num);
+ if (!st.ok()) {
+ LOG(WARNING) << st.to_string();
+ } else {
+ ctx->pulsar_info
+ ->partition_backlog[std::static_pointer_cast<PulsarDataConsumer>(consumer)->get_partition()] =
+ backlog_num;
+ }
+ }
+}
+
+void PulsarDataConsumerGroup::acknowledge_cumulative(std::shared_ptr<StreamLoadContext> ctx) {
+ for (auto& kv : ctx->pulsar_info->ack_offset) {
+ LOG(INFO) << "start do ack of message_id: " << kv.second
+ << "partition: " << kv.first;
+ for (auto& consumer : _consumers) {
+ // do ack
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->acknowledge_cumulative(kv.second);
+ }
+ LOG(INFO) << "finish do ack of message_id: " << kv.second
+ << "partition: " << kv.first;
+ }
+}
+
+const char* PulsarDataConsumerGroup::filter_invalid_prefix_of_json(const char* data, std::size_t size) {
Review Comment:
warning: method 'filter_invalid_prefix_of_json' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/routine_load/data_consumer_group.h:126:
```diff
- const char* filter_invalid_prefix_of_json(const char* data, std::size_t size);
+ static const char* filter_invalid_prefix_of_json(const char* data, std::size_t size);
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
+ int64_t received_rows = 0;
+ int64_t left_bytes = ctx->max_batch_size;
+
+ std::shared_ptr<io::PulsarConsumerPipe> pulsar_pipe = std::static_pointer_cast<io::PulsarConsumerPipe>(ctx->body_sink);
+
+ LOG(INFO) << "start consumer group: " << _grp_id << ". max time(ms): " << left_time
+ << ", batch size: " << left_bytes << ". " << ctx->brief();
+
+ // copy one
+ std::map<std::string, pulsar::MessageId> ack_offset = ctx->pulsar_info->ack_offset;
+
+ //improve performance
+ Status (io::PulsarConsumerPipe::*append_data)(const char* data, size_t size);
+ if (ctx->format == TFileFormatType::FORMAT_JSON) {
+ append_data = &io::PulsarConsumerPipe::append_json;
+ } else {
+ append_data = &io::PulsarConsumerPipe::append_with_line_delimiter;
+ }
+
+ MonotonicStopWatch watch;
+ watch.start();
+ bool eos = false;
+ while (true) {
+ if (eos || left_time <= 0 || left_bytes <= 0) {
+ LOG(INFO) << "consumer group done: " << _grp_id
+ << ". consume time(ms)=" << ctx->max_interval_s * 1000 - left_time
+ << ", received rows=" << received_rows << ", received bytes=" << ctx->max_batch_size - left_bytes
+ << ", eos: " << eos << ", left_time: " << left_time << ", left_bytes: " << left_bytes
+ << ", blocking get time(us): " << _queue.total_get_wait_time() / 1000
+ << ", blocking put time(us): " << _queue.total_put_wait_time() / 1000;
+
+ // cancel all consumers
+ for (auto& consumer : _consumers) {
+ static_cast<void>(consumer->cancel(ctx));
+ }
+
+ // sleep 3s for waitting consumer cancel done
+ std::this_thread::sleep_for(std::chrono::seconds(3));
+ // shutdown queue
+ _queue.shutdown();
+
+ // waiting all threads finished
+ _thread_pool.shutdown();
+ _thread_pool.join();
+
+ if (!result_st.ok()) {
+ // some consumers encounter errors, cancel this task
+ LOG(WARNING) << "Failed. Cancelled append msg to pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel(result_st.to_string());
+ return result_st;
+ }
+
+ if (left_bytes == ctx->max_batch_size) {
+ // nothing to be consumed, we have to cancel it, because
+ // we do not allow finishing stream load pipe without data
+ LOG(WARNING) << "Cancelled append msg to pulsar pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel("Cancelled");
+ return Status::InternalError("Cancelled");
+ } else {
+ DCHECK(left_bytes < ctx->max_batch_size);
+ static_cast<void>(pulsar_pipe->finish());
+ ctx->pulsar_info->ack_offset = std::move(ack_offset);
+ ctx->receive_bytes = ctx->max_batch_size - left_bytes;
+ get_backlog_nums(ctx);
+ acknowledge_cumulative(ctx);
+ return Status::OK();
+ }
+ }
+
+ pulsar::Message* msg;
+ bool res = _queue.blocking_get(&msg);
+ if (res) {
+ std::string partition = msg->getProperty("topicName");
+ pulsar::MessageId msg_id = msg->getMessageId();
+ std::size_t len = msg->getLength();
+
+ VLOG(3) << "get pulsar message:" << msg->getDataAsString()
+ << ", partition: " << partition << ", message id: " << msg_id
+ << ", len: " << len << ", size: " << msg->getDataAsString().size();
+
+ Status st = (pulsar_pipe.get()->*append_data)(msg->getDataAsString().c_str(),
+ msg->getDataAsString().size());
+ if (st.ok()) {
+ received_rows++;
+ // len of receive origin message from pulsar
+ left_bytes -= len;
+ if (ack_offset.find(partition) != ack_offset.end()) {
+ if (msg->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+ LOG(INFO) << "old message id: " << ack_offset[partition]
+ << ", new message id: " << msg_id;
+ }
+ if (ack_offset[partition] < msg_id) {
+ ack_offset[partition] = msg_id;
+ }
+ } else {
+ ack_offset[partition] = msg_id;
+ }
+ VLOG(3) << "consume partition" << partition << " - " << msg_id;
+ } else {
+ // failed to append this msg, we must stop
+ LOG(WARNING) << "failed to append msg to pipe. grp: " << _grp_id << ", errmsg=" << st.to_string();
+ eos = true;
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ if (result_st.ok()) {
+ result_st = st;
+ }
+ }
+ }
+ delete msg;
+ } else {
+ // queue is empty and shutdown
+ eos = true;
+ }
+
+ left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
+ }
+
+ return Status::OK();
+}
+
+void PulsarDataConsumerGroup::actual_consume(const std::shared_ptr<DataConsumer>& consumer,
+ BlockingQueue<pulsar::Message*>* queue, int64_t max_running_time_ms,
+ const ConsumeFinishCallback& cb) {
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->group_consume(queue, max_running_time_ms);
+ cb(st);
+}
+
+void PulsarDataConsumerGroup::get_backlog_nums(std::shared_ptr<StreamLoadContext> ctx) {
+ for (auto& consumer : _consumers) {
+ // get backlog num
+ int64_t backlog_num;
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->get_partition_backlog(&backlog_num);
+ if (!st.ok()) {
+ LOG(WARNING) << st.to_string();
+ } else {
+ ctx->pulsar_info
+ ->partition_backlog[std::static_pointer_cast<PulsarDataConsumer>(consumer)->get_partition()] =
+ backlog_num;
+ }
+ }
+}
+
+void PulsarDataConsumerGroup::acknowledge_cumulative(std::shared_ptr<StreamLoadContext> ctx) {
+ for (auto& kv : ctx->pulsar_info->ack_offset) {
+ LOG(INFO) << "start do ack of message_id: " << kv.second
+ << "partition: " << kv.first;
+ for (auto& consumer : _consumers) {
+ // do ack
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->acknowledge_cumulative(kv.second);
+ }
+ LOG(INFO) << "finish do ack of message_id: " << kv.second
+ << "partition: " << kv.first;
+ }
+}
+
+const char* PulsarDataConsumerGroup::filter_invalid_prefix_of_json(const char* data, std::size_t size) {
+ // first index of '{'
+ int first_left_curly_bracket_index = -1;
+ for (int i = 0; i < size; ++i) {
+ if (first_left_curly_bracket_index == -1 && data[i] == '{') {
+ first_left_curly_bracket_index = i;
+ if ( i+1 < size && data[i+1] == '{') {
+ first_left_curly_bracket_index = i + 1;
+ }
+ break;
+ }
+ }
+ if (first_left_curly_bracket_index >= 0) {
+ return data + first_left_curly_bracket_index;
+ } else {
+ return data;
+ }
Review Comment:
warning: do not use 'else' after 'return' [readability-else-after-return]
```suggestion
} return data;
```
##########
be/src/runtime/routine_load/data_consumer.cpp:
##########
@@ -452,4 +452,360 @@
return true;
}
+// init pulsar consumer will only set common configs
+Status PulsarDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_init) {
+ // this consumer has already been initialized.
+ return Status::OK();
+ }
+
+ pulsar::ClientConfiguration config;
+ for (auto& item : ctx->pulsar_info->properties) {
+ if (item.first == "auth.token") {
+ config.setAuth(pulsar::AuthToken::createWithToken(item.second));
+ } else {
+ LOG(WARNING) << "Config " << item.first << " not supported for now.";
+ }
+
+ _custom_properties.emplace(item.first, item.second);
+ }
+
+ _p_client = new pulsar::Client(_service_url, config);
+
+ VLOG(3) << "finished to init pulsar consumer. " << ctx->brief();
+
+ _init = true;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::assign_partition(const std::string& partition, std::shared_ptr<StreamLoadContext> ctx,
+ int64_t initial_position) {
+ DCHECK(_p_client);
+
+ std::stringstream ss;
+ ss << "consumer: " << _id << ", grp: " << _grp_id << " topic: " << _topic
+ << ", subscription: " << _subscription << ", initial_position: " << initial_position;
+ LOG(INFO) << ss.str();
+
+ // do subscribe
+ pulsar::ConsumerConfiguration conf;
+ conf.setConsumerType(pulsar::ConsumerExclusive);
+
+ pulsar::Result result;
+ result = _p_client->subscribe(partition, _subscription, conf, _p_consumer);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to create pulsar consumer: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to create pulsar consumer: " +
+ std::string(pulsar::strResult(result)));
+ }
+
+ if (initial_position == InitialPosition::LATEST || initial_position == InitialPosition::EARLIEST) {
+ pulsar::InitialPosition p_initial_position = initial_position == InitialPosition::LATEST
+ ? pulsar::InitialPosition::InitialPositionLatest
+ : pulsar::InitialPosition::InitialPositionEarliest;
+ result = _p_consumer.seek(p_initial_position);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to reset the subscription: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to reset the subscription: " +
+ std::string(pulsar::strResult(result)));
+ }
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::group_consume(BlockingQueue<pulsar::Message*>* queue, int64_t max_running_time_ms) {
+ _last_visit_time = time(nullptr);
+ int64_t left_time = max_running_time_ms;
+ LOG(INFO) << "start pulsar consumer: " << _id << ", grp: " << _grp_id << ", max running time(ms): " << left_time;
+
+ int64_t received_rows = 0;
+ int64_t put_rows = 0;
+ Status st = Status::OK();
+ pulsar::MessageId msg_id;
+ std::string message_str;
+ MonotonicStopWatch consumer_watch;
+ MonotonicStopWatch watch;
+ watch.start();
+ while (true) {
+ {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_cancelled) {
+ break;
+ }
+ }
+
+ if (left_time <= 0) {
+ break;
+ }
+
+ bool done = false;
+ auto msg = std::make_unique<pulsar::Message>();
+ std::vector<const char*> rows;
+ std::string filter_data;
+ pulsar::Message* new_msg;
+ // consume 1 message at a time
+ consumer_watch.start();
+ pulsar::Result res = _p_consumer.receive(*(msg.get()), 30000 /* timeout, ms */);
+ consumer_watch.stop();
+ switch (res) {
+ case pulsar::ResultOk:
+ //filter invalid prefix of json
+ filter_data = substring_prefix_json(msg.get()->getDataAsString());
+ rows = convert_rows(filter_data.c_str());
+ msg_id = msg.get()->getMessageId();
+ message_str = msg.get()->getDataAsString();
+ if (received_rows == 0) {
+ LOG(INFO) << "receive first pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ }
+// if (msg.get()->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+// LOG(INFO) << "receive pulsar message "
+// << ", len: " << msg.get()->getLength()
+// << ", message id: " << msg_id
+// << ", pulsar consumer: " << _id
+// << ", grp: " << _grp_id
+// << ", rows size: " << rows.size();
+// }
+ for (const char* row : rows) {
+ pulsar::MessageBuilder messageBuilder;
+ size_t row_len = len_of_actual_data(row);
+ messageBuilder.setContent(row, row_len);
+ messageBuilder.setProperty("topicName",msg.get()->getTopicName());
+ new_msg = new pulsar::Message(messageBuilder.build());
+ new_msg->setMessageId(msg_id);
+
+ if (new_msg->getDataAsString().find("{\"") == std::string::npos) {
+ // ignore msg with length 0.
+ // put empty msg into queue will cause the load process shutting down.
+ LOG(INFO) << "pass error message: " << new_msg->getDataAsString();
+ break;
+ } else if (!queue->blocking_put(&(*new_msg))) {
+ // queue is shutdown
+ LOG(INFO) << "queue is shutdown, failed to blocking put" << new_msg->getDataAsString()
+ << ", len: " << new_msg->getLength()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ done = true;
+ } else {
+ ++put_rows;
+ }
+ ++received_rows;
+ }
+ delete msg.get();
+ msg.release(); // release the ownership, msg will be deleted after being processed
+ for (const char* row : rows) {
+ delete[] row;
+ }
+ rows.clear();
+ break;
+ case pulsar::ResultTimeout:
+ // leave the status as OK, because this may happened
+ // if there is no data in pulsar.
+ LOG(INFO) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume timeout.";
+ break;
+ default:
+ LOG(WARNING) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume failed: "
+ << ", errmsg: " << res;
+ done = true;
+ st = Status::InternalError(pulsar::strResult(res));
+ break;
+ }
+
+ left_time = max_running_time_ms - watch.elapsed_time() / 1000 / 1000;
+ if (done) {
+ break;
+ }
+ }
+
+ LOG(INFO) << "receive last pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id;
+
+// LOG(INFO) << "start do ack of msg_id :" << msg_id;
+// pulsar::Result ack = _p_consumer.acknowledgeCumulative(msg_id);
+// if (ack != pulsar::ResultOk) {
+// LOG(WARNING) << "failed do ack of msg_id :" << msg_id << ", consumer : " << _id;
+// } else {
+// LOG(INFO) << "finish do ack of msg_id :" << msg_id << ", consumer : " << _id;
+// }
+
+ LOG(INFO) << "pulsar consume done: " << _id << ", grp: " << _grp_id << ". cancelled: " << _cancelled
+ << ", left time(ms): " << left_time << ", total cost(ms): " << watch.elapsed_time() / 1000 / 1000
+ << ", consume cost(ms): " << consumer_watch.elapsed_time() / 1000 / 1000
+ << ", received rows: " << received_rows << ", put rows: " << put_rows
+ << ", last message id: " << msg_id;
+
+ return st;
+}
+
+const std::string& PulsarDataConsumer::get_partition() {
+ _last_visit_time = time(nullptr);
+ return _p_consumer.getTopic();
+}
+
+Status PulsarDataConsumer::get_partition_backlog(int64_t* backlog) {
+ _last_visit_time = time(nullptr);
+ pulsar::BrokerConsumerStats broker_consumer_stats;
+ pulsar::Result result = _p_consumer.getBrokerConsumerStats(broker_consumer_stats);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "Failed to get broker consumer stats: "
+ << ", err: " << result;
+ return Status::InternalError("Failed to get broker consumer stats: " + std::string(pulsar::strResult(result)));
+ }
+ *backlog = broker_consumer_stats.getMsgBacklog();
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::get_topic_partition(std::vector<std::string>* partitions) {
+ _last_visit_time = time(nullptr);
+ pulsar::Result result = _p_client->getPartitionsForTopic(_topic, *partitions);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "Failed to get partitions for topic: " << _topic << ", err: " << result;
+ return Status::InternalError("Failed to get partitions for topic: " + std::string(pulsar::strResult(result)));
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::cancel(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (!_init) {
+ return Status::InternalError("consumer is not initialized");
+ }
+
+ _cancelled = true;
+ LOG(INFO) << "pulsar consumer cancelled. " << _id;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::reset() {
+ std::unique_lock<std::mutex> l(_lock);
+ _cancelled = false;
+ _p_consumer.close();
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::acknowledge_cumulative(pulsar::MessageId& message_id) {
+ pulsar::Result res = _p_consumer.acknowledgeCumulative(message_id);
+ if (res != pulsar::ResultOk) {
+ std::stringstream ss;
+ ss << "failed to acknowledge pulsar message : " << res;
+ LOG(WARNING) << "failed to acknowledge pulsar message : " << res
+ << ",pulsar consumer :" << _id
+ << ",pulsar group :" << _grp_id;
+ return Status::InternalError(ss.str());
+ }
+ LOG(INFO) << "message_id :" << message_id
+ << ",pulsar consumer :" << _id
+ << ",pulsar group :" << _grp_id;
+ return Status::OK();
+}
+
+bool PulsarDataConsumer::match(std::shared_ptr<StreamLoadContext> ctx) {
+ if (ctx->load_src_type != TLoadSourceType::PULSAR) {
+ return false;
+ }
+ if (_service_url != ctx->pulsar_info->service_url || _topic != ctx->pulsar_info->topic ||
+ _subscription != ctx->pulsar_info->subscription) {
+ return false;
+ }
+ // check properties
+ if (_custom_properties.size() != ctx->pulsar_info->properties.size()) {
+ return false;
+ }
+ for (auto& item : ctx->pulsar_info->properties) {
+ auto it = _custom_properties.find(item.first);
+ if (it == _custom_properties.end() || it->second != item.second) {
+ return false;
+ }
+ }
+
+ return true;
+}
+
+std::string PulsarDataConsumer::substring_prefix_json(std::string data) {
+ // 找到以 "{" 开头的位置
+ size_t startPos = data.find("{\"");
+
+ // 如果找到了，则截取并返回子字符串
+ if (startPos != std::string::npos) {
+ return data.substr(startPos);
+ } else {
+ return data;
+ }
+}
+
+size_t PulsarDataConsumer::len_of_actual_data(const char* data) {
Review Comment:
warning: do not use 'else' after 'return' [readability-else-after-return]
```suggestion
os); return data;
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
+ int64_t received_rows = 0;
+ int64_t left_bytes = ctx->max_batch_size;
+
+ std::shared_ptr<io::PulsarConsumerPipe> pulsar_pipe = std::static_pointer_cast<io::PulsarConsumerPipe>(ctx->body_sink);
+
+ LOG(INFO) << "start consumer group: " << _grp_id << ". max time(ms): " << left_time
+ << ", batch size: " << left_bytes << ". " << ctx->brief();
+
+ // copy one
+ std::map<std::string, pulsar::MessageId> ack_offset = ctx->pulsar_info->ack_offset;
+
+ //improve performance
+ Status (io::PulsarConsumerPipe::*append_data)(const char* data, size_t size);
+ if (ctx->format == TFileFormatType::FORMAT_JSON) {
+ append_data = &io::PulsarConsumerPipe::append_json;
+ } else {
+ append_data = &io::PulsarConsumerPipe::append_with_line_delimiter;
+ }
+
+ MonotonicStopWatch watch;
+ watch.start();
+ bool eos = false;
+ while (true) {
+ if (eos || left_time <= 0 || left_bytes <= 0) {
+ LOG(INFO) << "consumer group done: " << _grp_id
+ << ". consume time(ms)=" << ctx->max_interval_s * 1000 - left_time
+ << ", received rows=" << received_rows << ", received bytes=" << ctx->max_batch_size - left_bytes
+ << ", eos: " << eos << ", left_time: " << left_time << ", left_bytes: " << left_bytes
+ << ", blocking get time(us): " << _queue.total_get_wait_time() / 1000
+ << ", blocking put time(us): " << _queue.total_put_wait_time() / 1000;
+
+ // cancel all consumers
+ for (auto& consumer : _consumers) {
+ static_cast<void>(consumer->cancel(ctx));
+ }
+
+ // sleep 3s for waitting consumer cancel done
+ std::this_thread::sleep_for(std::chrono::seconds(3));
+ // shutdown queue
+ _queue.shutdown();
+
+ // waiting all threads finished
+ _thread_pool.shutdown();
+ _thread_pool.join();
+
+ if (!result_st.ok()) {
+ // some consumers encounter errors, cancel this task
+ LOG(WARNING) << "Failed. Cancelled append msg to pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel(result_st.to_string());
+ return result_st;
+ }
+
+ if (left_bytes == ctx->max_batch_size) {
+ // nothing to be consumed, we have to cancel it, because
+ // we do not allow finishing stream load pipe without data
+ LOG(WARNING) << "Cancelled append msg to pulsar pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel("Cancelled");
+ return Status::InternalError("Cancelled");
+ } else {
+ DCHECK(left_bytes < ctx->max_batch_size);
+ static_cast<void>(pulsar_pipe->finish());
+ ctx->pulsar_info->ack_offset = std::move(ack_offset);
+ ctx->receive_bytes = ctx->max_batch_size - left_bytes;
+ get_backlog_nums(ctx);
+ acknowledge_cumulative(ctx);
+ return Status::OK();
+ }
+ }
+
+ pulsar::Message* msg;
+ bool res = _queue.blocking_get(&msg);
+ if (res) {
+ std::string partition = msg->getProperty("topicName");
+ pulsar::MessageId msg_id = msg->getMessageId();
+ std::size_t len = msg->getLength();
+
+ VLOG(3) << "get pulsar message:" << msg->getDataAsString()
+ << ", partition: " << partition << ", message id: " << msg_id
+ << ", len: " << len << ", size: " << msg->getDataAsString().size();
+
+ Status st = (pulsar_pipe.get()->*append_data)(msg->getDataAsString().c_str(),
+ msg->getDataAsString().size());
+ if (st.ok()) {
+ received_rows++;
+ // len of receive origin message from pulsar
+ left_bytes -= len;
+ if (ack_offset.find(partition) != ack_offset.end()) {
+ if (msg->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+ LOG(INFO) << "old message id: " << ack_offset[partition]
+ << ", new message id: " << msg_id;
+ }
+ if (ack_offset[partition] < msg_id) {
+ ack_offset[partition] = msg_id;
+ }
+ } else {
+ ack_offset[partition] = msg_id;
+ }
+ VLOG(3) << "consume partition" << partition << " - " << msg_id;
+ } else {
+ // failed to append this msg, we must stop
+ LOG(WARNING) << "failed to append msg to pipe. grp: " << _grp_id << ", errmsg=" << st.to_string();
+ eos = true;
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ if (result_st.ok()) {
+ result_st = st;
+ }
+ }
+ }
+ delete msg;
+ } else {
+ // queue is empty and shutdown
+ eos = true;
+ }
+
+ left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
+ }
+
+ return Status::OK();
+}
+
+void PulsarDataConsumerGroup::actual_consume(const std::shared_ptr<DataConsumer>& consumer,
+ BlockingQueue<pulsar::Message*>* queue, int64_t max_running_time_ms,
+ const ConsumeFinishCallback& cb) {
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->group_consume(queue, max_running_time_ms);
+ cb(st);
+}
+
+void PulsarDataConsumerGroup::get_backlog_nums(std::shared_ptr<StreamLoadContext> ctx) {
+ for (auto& consumer : _consumers) {
+ // get backlog num
+ int64_t backlog_num;
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->get_partition_backlog(&backlog_num);
+ if (!st.ok()) {
+ LOG(WARNING) << st.to_string();
+ } else {
+ ctx->pulsar_info
+ ->partition_backlog[std::static_pointer_cast<PulsarDataConsumer>(consumer)->get_partition()] =
+ backlog_num;
+ }
+ }
+}
+
+void PulsarDataConsumerGroup::acknowledge_cumulative(std::shared_ptr<StreamLoadContext> ctx) {
+ for (auto& kv : ctx->pulsar_info->ack_offset) {
+ LOG(INFO) << "start do ack of message_id: " << kv.second
+ << "partition: " << kv.first;
+ for (auto& consumer : _consumers) {
+ // do ack
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->acknowledge_cumulative(kv.second);
+ }
+ LOG(INFO) << "finish do ack of message_id: " << kv.second
+ << "partition: " << kv.first;
+ }
+}
+
+const char* PulsarDataConsumerGroup::filter_invalid_prefix_of_json(const char* data, std::size_t size) {
+ // first index of '{'
+ int first_left_curly_bracket_index = -1;
+ for (int i = 0; i < size; ++i) {
+ if (first_left_curly_bracket_index == -1 && data[i] == '{') {
+ first_left_curly_bracket_index = i;
+ if ( i+1 < size && data[i+1] == '{') {
+ first_left_curly_bracket_index = i + 1;
+ }
+ break;
+ }
+ }
+ if (first_left_curly_bracket_index >= 0) {
+ return data + first_left_curly_bracket_index;
+ } else {
+ return data;
+ }
+}
+
+size_t PulsarDataConsumerGroup::len_of_actual_data(const char* data) {
Review Comment:
warning: method 'len_of_actual_data' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/routine_load/data_consumer_group.h:128:
```diff
- size_t len_of_actual_data(const char* data);
+ static size_t len_of_actual_data(const char* data);
```
##########
be/src/runtime/routine_load/routine_load_task_executor.cpp:
##########
@@ -173,6 +173,93 @@ Status RoutineLoadTaskExecutor::get_kafka_latest_offsets_for_partitions(
return st;
}
+Status RoutineLoadTaskExecutor::get_pulsar_partition_meta(const PPulsarMetaProxyRequest& request,
Review Comment:
warning: method 'get_pulsar_partition_meta' can be made static [readability-convert-member-functions-to-static]
```suggestion
static Status RoutineLoadTaskExecutor::get_pulsar_partition_meta(const PPulsarMetaProxyRequest& request,
```
##########
be/src/runtime/routine_load/data_consumer_pool.cpp:
##########
@@ -77,31 +81,59 @@ Status DataConsumerPool::get_consumer(std::shared_ptr<StreamLoadContext> ctx,
Status DataConsumerPool::get_consumer_grp(std::shared_ptr<StreamLoadContext> ctx,
std::shared_ptr<DataConsumerGroup>* ret) {
- if (ctx->load_src_type != TLoadSourceType::KAFKA) {
+ if (ctx->load_src_type != TLoadSourceType::KAFKA && ctx->load_src_type != TLoadSourceType::PULSAR) {
return Status::InternalError(
- "PAUSE: Currently only support consumer group for Kafka data source");
+ "PAUSE: Currently only support consumer group for Kafka or Pulsar data source");
}
- DCHECK(ctx->kafka_info);
- if (ctx->kafka_info->begin_offset.size() == 0) {
- return Status::InternalError("PAUSE: The size of begin_offset of task should not be 0.");
- }
+ if (ctx->load_src_type == TLoadSourceType::KAFKA) {
+ DCHECK(ctx->kafka_info);
- std::shared_ptr<KafkaDataConsumerGroup> grp = std::make_shared<KafkaDataConsumerGroup>();
+ if (ctx->kafka_info->begin_offset.size() == 0) {
+ return Status::InternalError("PAUSE: The size of begin_offset of task should not be 0.");
+ }
- // one data consumer group contains at least one data consumers.
- int max_consumer_num = config::max_consumer_num_per_group;
- size_t consumer_num = std::min((size_t)max_consumer_num, ctx->kafka_info->begin_offset.size());
+ std::shared_ptr<KafkaDataConsumerGroup> grp = std::make_shared<KafkaDataConsumerGroup>();
- for (int i = 0; i < consumer_num; ++i) {
- std::shared_ptr<DataConsumer> consumer;
- RETURN_IF_ERROR(get_consumer(ctx, &consumer));
- grp->add_consumer(consumer);
- }
+ // one data consumer group contains at least one data consumers.
+ int max_consumer_num = config::max_consumer_num_per_group;
+ size_t consumer_num = std::min((size_t)max_consumer_num, ctx->kafka_info->begin_offset.size());
- LOG(INFO) << "get consumer group " << grp->grp_id() << " with " << consumer_num << " consumers";
- *ret = grp;
- return Status::OK();
+ for (int i = 0; i < consumer_num; ++i) {
+ std::shared_ptr<DataConsumer> consumer;
+ RETURN_IF_ERROR(get_consumer(ctx, &consumer));
+ grp->add_consumer(consumer);
+ }
+
+ LOG(INFO) << "get consumer group " << grp->grp_id() << " with " << consumer_num << " consumers";
+ *ret = grp;
+ return Status::OK();
+ } else {
+ DCHECK(ctx->pulsar_info);
+ DCHECK_GE(ctx->pulsar_info->partitions.size(), 1);
+
+ // Cumulative acknowledge is not supported for multiple topic subscribtion,
+ // so one consumer can only subscribe one topic/partition
+ int max_consumer_num = config::max_pulsar_consumer_num_per_group;
+ if (max_consumer_num < ctx->pulsar_info->partitions.size()) {
+ return Status::InternalError(
+ "PAUSE: Partition num is more than max consumer num in one data consumer group on some BEs, please "
+ "increase max_pulsar_consumer_num_per_group from BE side or just add more BEs");
+ }
+ size_t consumer_num = ctx->pulsar_info->partitions.size();
+
+ std::shared_ptr<PulsarDataConsumerGroup> grp = std::make_shared<PulsarDataConsumerGroup>();
+
+ for (int i = 0; i < consumer_num; ++i) {
+ std::shared_ptr<DataConsumer> consumer;
+ RETURN_IF_ERROR(get_consumer(ctx, &consumer));
+ grp->add_consumer(consumer);
+ }
+
+ LOG(INFO) << "get consumer group " << grp->grp_id() << " with " << consumer_num << " consumers";
+ *ret = grp;
+ return Status::OK();
+ }
Review Comment:
warning: do not use 'else' after 'return' [readability-else-after-return]
```suggestion
} DCHECK(ctx->pulsar_info);
DCHECK_GE(ctx->pulsar_info->partitions.size(), 1);
// Cumulative acknowledge is not supported for multiple topic subscribtion,
// so one consumer can only subscribe one topic/partition
int max_consumer_num = config::max_pulsar_consumer_num_per_group;
if (max_consumer_num < ctx->pulsar_info->partitions.size()) {
return Status::InternalError(
"PAUSE: Partition num is more than max consumer num in one data consumer group on some BEs, please "
"increase max_pulsar_consumer_num_per_group from BE side or just add more BEs");
}
size_t consumer_num = ctx->pulsar_info->partitions.size();
std::shared_ptr<PulsarDataConsumerGroup> grp = std::make_shared<PulsarDataConsumerGroup>();
for (int i = 0; i < consumer_num; ++i) {
std::shared_ptr<DataConsumer> consumer;
RETURN_IF_ERROR(get_consumer(ctx, &consumer));
grp->add_consumer(consumer);
}
LOG(INFO) << "get consumer group " << grp->grp_id() << " with " << consumer_num << " consumers";
*ret = grp;
return Status::OK();
```
##########
be/src/runtime/routine_load/data_consumer_group.h:
##########
@@ -108,4 +106,34 @@ class KafkaDataConsumerGroup : public DataConsumerGroup {
BlockingQueue<RdKafka::Message*> _queue;
};
+// for pulsar
+class PulsarDataConsumerGroup : public DataConsumerGroup {
+public:
+ PulsarDataConsumerGroup() : DataConsumerGroup(), _queue(3000) {}
Review Comment:
warning: initializer for base class 'doris::DataConsumerGroup' is redundant [readability-redundant-member-init]
```suggestion
PulsarDataConsumerGroup() : , _queue(3000) {}
```
##########
be/src/runtime/routine_load/data_consumer.cpp:
##########
@@ -452,4 +452,360 @@
return true;
}
+// init pulsar consumer will only set common configs
+Status PulsarDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_init) {
+ // this consumer has already been initialized.
+ return Status::OK();
+ }
+
+ pulsar::ClientConfiguration config;
+ for (auto& item : ctx->pulsar_info->properties) {
+ if (item.first == "auth.token") {
+ config.setAuth(pulsar::AuthToken::createWithToken(item.second));
+ } else {
+ LOG(WARNING) << "Config " << item.first << " not supported for now.";
+ }
+
+ _custom_properties.emplace(item.first, item.second);
+ }
+
+ _p_client = new pulsar::Client(_service_url, config);
+
+ VLOG(3) << "finished to init pulsar consumer. " << ctx->brief();
+
+ _init = true;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::assign_partition(const std::string& partition, std::shared_ptr<StreamLoadContext> ctx,
+ int64_t initial_position) {
+ DCHECK(_p_client);
+
+ std::stringstream ss;
+ ss << "consumer: " << _id << ", grp: " << _grp_id << " topic: " << _topic
+ << ", subscription: " << _subscription << ", initial_position: " << initial_position;
+ LOG(INFO) << ss.str();
+
+ // do subscribe
+ pulsar::ConsumerConfiguration conf;
+ conf.setConsumerType(pulsar::ConsumerExclusive);
+
+ pulsar::Result result;
+ result = _p_client->subscribe(partition, _subscription, conf, _p_consumer);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to create pulsar consumer: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to create pulsar consumer: " +
+ std::string(pulsar::strResult(result)));
+ }
+
+ if (initial_position == InitialPosition::LATEST || initial_position == InitialPosition::EARLIEST) {
+ pulsar::InitialPosition p_initial_position = initial_position == InitialPosition::LATEST
+ ? pulsar::InitialPosition::InitialPositionLatest
+ : pulsar::InitialPosition::InitialPositionEarliest;
+ result = _p_consumer.seek(p_initial_position);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "PAUSE: failed to reset the subscription: " << ctx->brief(true) << ", err: " << result;
+ return Status::InternalError("PAUSE: failed to reset the subscription: " +
+ std::string(pulsar::strResult(result)));
+ }
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::group_consume(BlockingQueue<pulsar::Message*>* queue, int64_t max_running_time_ms) {
+ _last_visit_time = time(nullptr);
+ int64_t left_time = max_running_time_ms;
+ LOG(INFO) << "start pulsar consumer: " << _id << ", grp: " << _grp_id << ", max running time(ms): " << left_time;
+
+ int64_t received_rows = 0;
+ int64_t put_rows = 0;
+ Status st = Status::OK();
+ pulsar::MessageId msg_id;
+ std::string message_str;
+ MonotonicStopWatch consumer_watch;
+ MonotonicStopWatch watch;
+ watch.start();
+ while (true) {
+ {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_cancelled) {
+ break;
+ }
+ }
+
+ if (left_time <= 0) {
+ break;
+ }
+
+ bool done = false;
+ auto msg = std::make_unique<pulsar::Message>();
+ std::vector<const char*> rows;
+ std::string filter_data;
+ pulsar::Message* new_msg;
+ // consume 1 message at a time
+ consumer_watch.start();
+ pulsar::Result res = _p_consumer.receive(*(msg.get()), 30000 /* timeout, ms */);
+ consumer_watch.stop();
+ switch (res) {
+ case pulsar::ResultOk:
+ //filter invalid prefix of json
+ filter_data = substring_prefix_json(msg.get()->getDataAsString());
+ rows = convert_rows(filter_data.c_str());
+ msg_id = msg.get()->getMessageId();
+ message_str = msg.get()->getDataAsString();
+ if (received_rows == 0) {
+ LOG(INFO) << "receive first pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ }
+// if (msg.get()->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+// LOG(INFO) << "receive pulsar message "
+// << ", len: " << msg.get()->getLength()
+// << ", message id: " << msg_id
+// << ", pulsar consumer: " << _id
+// << ", grp: " << _grp_id
+// << ", rows size: " << rows.size();
+// }
+ for (const char* row : rows) {
+ pulsar::MessageBuilder messageBuilder;
+ size_t row_len = len_of_actual_data(row);
+ messageBuilder.setContent(row, row_len);
+ messageBuilder.setProperty("topicName",msg.get()->getTopicName());
+ new_msg = new pulsar::Message(messageBuilder.build());
+ new_msg->setMessageId(msg_id);
+
+ if (new_msg->getDataAsString().find("{\"") == std::string::npos) {
+ // ignore msg with length 0.
+ // put empty msg into queue will cause the load process shutting down.
+ LOG(INFO) << "pass error message: " << new_msg->getDataAsString();
+ break;
+ } else if (!queue->blocking_put(&(*new_msg))) {
+ // queue is shutdown
+ LOG(INFO) << "queue is shutdown, failed to blocking put" << new_msg->getDataAsString()
+ << ", len: " << new_msg->getLength()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id
+ << ", rows size: " << rows.size();
+ done = true;
+ } else {
+ ++put_rows;
+ }
+ ++received_rows;
+ }
+ delete msg.get();
+ msg.release(); // release the ownership, msg will be deleted after being processed
+ for (const char* row : rows) {
+ delete[] row;
+ }
+ rows.clear();
+ break;
+ case pulsar::ResultTimeout:
+ // leave the status as OK, because this may happened
+ // if there is no data in pulsar.
+ LOG(INFO) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume timeout.";
+ break;
+ default:
+ LOG(WARNING) << "pulsar consumer"
+ << "[" << _id << "]"
+ << " consume failed: "
+ << ", errmsg: " << res;
+ done = true;
+ st = Status::InternalError(pulsar::strResult(res));
+ break;
+ }
+
+ left_time = max_running_time_ms - watch.elapsed_time() / 1000 / 1000;
+ if (done) {
+ break;
+ }
+ }
+
+ LOG(INFO) << "receive last pulsar message: " << message_str
+ << ", len: " << message_str.size()
+ << ", message id: " << msg_id
+ << ", pulsar consumer: " << _id
+ << ", grp: " << _grp_id;
+
+// LOG(INFO) << "start do ack of msg_id :" << msg_id;
+// pulsar::Result ack = _p_consumer.acknowledgeCumulative(msg_id);
+// if (ack != pulsar::ResultOk) {
+// LOG(WARNING) << "failed do ack of msg_id :" << msg_id << ", consumer : " << _id;
+// } else {
+// LOG(INFO) << "finish do ack of msg_id :" << msg_id << ", consumer : " << _id;
+// }
+
+ LOG(INFO) << "pulsar consume done: " << _id << ", grp: " << _grp_id << ". cancelled: " << _cancelled
+ << ", left time(ms): " << left_time << ", total cost(ms): " << watch.elapsed_time() / 1000 / 1000
+ << ", consume cost(ms): " << consumer_watch.elapsed_time() / 1000 / 1000
+ << ", received rows: " << received_rows << ", put rows: " << put_rows
+ << ", last message id: " << msg_id;
+
+ return st;
+}
+
+const std::string& PulsarDataConsumer::get_partition() {
+ _last_visit_time = time(nullptr);
+ return _p_consumer.getTopic();
+}
+
+Status PulsarDataConsumer::get_partition_backlog(int64_t* backlog) {
+ _last_visit_time = time(nullptr);
+ pulsar::BrokerConsumerStats broker_consumer_stats;
+ pulsar::Result result = _p_consumer.getBrokerConsumerStats(broker_consumer_stats);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "Failed to get broker consumer stats: "
+ << ", err: " << result;
+ return Status::InternalError("Failed to get broker consumer stats: " + std::string(pulsar::strResult(result)));
+ }
+ *backlog = broker_consumer_stats.getMsgBacklog();
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::get_topic_partition(std::vector<std::string>* partitions) {
+ _last_visit_time = time(nullptr);
+ pulsar::Result result = _p_client->getPartitionsForTopic(_topic, *partitions);
+ if (result != pulsar::ResultOk) {
+ LOG(WARNING) << "Failed to get partitions for topic: " << _topic << ", err: " << result;
+ return Status::InternalError("Failed to get partitions for topic: " + std::string(pulsar::strResult(result)));
+ }
+
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::cancel(std::shared_ptr<StreamLoadContext> ctx) {
+ std::unique_lock<std::mutex> l(_lock);
+ if (!_init) {
+ return Status::InternalError("consumer is not initialized");
+ }
+
+ _cancelled = true;
+ LOG(INFO) << "pulsar consumer cancelled. " << _id;
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::reset() {
+ std::unique_lock<std::mutex> l(_lock);
+ _cancelled = false;
+ _p_consumer.close();
+ return Status::OK();
+}
+
+Status PulsarDataConsumer::acknowledge_cumulative(pulsar::MessageId& message_id) {
+ pulsar::Result res = _p_consumer.acknowledgeCumulative(message_id);
+ if (res != pulsar::ResultOk) {
+ std::stringstream ss;
+ ss << "failed to acknowledge pulsar message : " << res;
+ LOG(WARNING) << "failed to acknowledge pulsar message : " << res
+ << ",pulsar consumer :" << _id
+ << ",pulsar group :" << _grp_id;
+ return Status::InternalError(ss.str());
+ }
+ LOG(INFO) << "message_id :" << message_id
+ << ",pulsar consumer :" << _id
+ << ",pulsar group :" << _grp_id;
+ return Status::OK();
+}
+
+bool PulsarDataConsumer::match(std::shared_ptr<StreamLoadContext> ctx) {
+ if (ctx->load_src_type != TLoadSourceType::PULSAR) {
+ return false;
+ }
+ if (_service_url != ctx->pulsar_info->service_url || _topic != ctx->pulsar_info->topic ||
+ _subscription != ctx->pulsar_info->subscription) {
+ return false;
+ }
+ // check properties
+ if (_custom_properties.size() != ctx->pulsar_info->properties.size()) {
+ return false;
+ }
+ for (auto& item : ctx->pulsar_info->properties) {
+ auto it = _custom_properties.find(item.first);
+ if (it == _custom_properties.end() || it->second != item.second) {
+ return false;
+ }
+ }
+
+ return true;
+}
+
+std::string PulsarDataConsumer::substring_prefix_json(std::string data) {
Review Comment:
warning: method 'substring_prefix_json' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/routine_load/data_consumer.h:210:
```diff
- std::string substring_prefix_json(std::string data);
+ static std::string substring_prefix_json(std::string data);
```
##########
be/src/runtime/stream_load/stream_load_context.h:
##########
@@ -87,6 +88,41 @@ class KafkaLoadInfo {
std::map<std::string, std::string> properties;
};
+// pulsar related info
+class PulsarLoadInfo {
+public:
+ explicit PulsarLoadInfo(const TPulsarLoadInfo& t_info)
+ : service_url(t_info.service_url),
+ topic(t_info.topic),
+ subscription(t_info.subscription),
+ partitions(t_info.partitions),
+ properties(t_info.properties) {
+ if (t_info.__isset.initial_positions) {
+ initial_positions = t_info.initial_positions;
+ }
+ }
+
+ void clear_backlog() {
+ // clear the backlog
+ partition_backlog.clear();
+ }
+
+public:
Review Comment:
warning: redundant access specifier has the same accessibility as the previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
<details>
<summary>Additional context</summary>
**be/src/runtime/stream_load/stream_load_context.h:92:** previously declared here
```cpp
public:
^
```
</details>
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
Review Comment:
warning: 1000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
int64_t left_time = ctx->max_interval_s * 1000;
^
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@ void KafkaDataConsumerGroup::actual_consume(std::shared_ptr<DataConsumer> consum
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
Review Comment:
warning: method 'assign_topic_partitions' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/routine_load/data_consumer_group.h:117:
```diff
- Status assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx);
+ static Status assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx);
```
##########
be/src/runtime/routine_load/data_consumer.h:
##########
@@ -165,4 +167,61 @@ class KafkaDataConsumer : public DataConsumer {
RdKafka::KafkaConsumer* _k_consumer = nullptr;
};
+class PulsarDataConsumer : public DataConsumer {
+public:
+ PulsarDataConsumer(std::shared_ptr<StreamLoadContext> ctx)
+ : DataConsumer(),
Review Comment:
warning: initializer for base class 'doris::DataConsumer' is redundant [readability-redundant-member-init]
```suggestion
: ,
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
+ int64_t received_rows = 0;
+ int64_t left_bytes = ctx->max_batch_size;
+
+ std::shared_ptr<io::PulsarConsumerPipe> pulsar_pipe = std::static_pointer_cast<io::PulsarConsumerPipe>(ctx->body_sink);
+
+ LOG(INFO) << "start consumer group: " << _grp_id << ". max time(ms): " << left_time
+ << ", batch size: " << left_bytes << ". " << ctx->brief();
+
+ // copy one
+ std::map<std::string, pulsar::MessageId> ack_offset = ctx->pulsar_info->ack_offset;
+
+ //improve performance
+ Status (io::PulsarConsumerPipe::*append_data)(const char* data, size_t size);
+ if (ctx->format == TFileFormatType::FORMAT_JSON) {
+ append_data = &io::PulsarConsumerPipe::append_json;
+ } else {
+ append_data = &io::PulsarConsumerPipe::append_with_line_delimiter;
+ }
+
+ MonotonicStopWatch watch;
+ watch.start();
+ bool eos = false;
+ while (true) {
+ if (eos || left_time <= 0 || left_bytes <= 0) {
+ LOG(INFO) << "consumer group done: " << _grp_id
+ << ". consume time(ms)=" << ctx->max_interval_s * 1000 - left_time
+ << ", received rows=" << received_rows << ", received bytes=" << ctx->max_batch_size - left_bytes
+ << ", eos: " << eos << ", left_time: " << left_time << ", left_bytes: " << left_bytes
+ << ", blocking get time(us): " << _queue.total_get_wait_time() / 1000
+ << ", blocking put time(us): " << _queue.total_put_wait_time() / 1000;
+
+ // cancel all consumers
+ for (auto& consumer : _consumers) {
+ static_cast<void>(consumer->cancel(ctx));
+ }
+
+ // sleep 3s for waitting consumer cancel done
+ std::this_thread::sleep_for(std::chrono::seconds(3));
+ // shutdown queue
+ _queue.shutdown();
+
+ // waiting all threads finished
+ _thread_pool.shutdown();
+ _thread_pool.join();
+
+ if (!result_st.ok()) {
+ // some consumers encounter errors, cancel this task
+ LOG(WARNING) << "Failed. Cancelled append msg to pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel(result_st.to_string());
+ return result_st;
+ }
+
+ if (left_bytes == ctx->max_batch_size) {
+ // nothing to be consumed, we have to cancel it, because
+ // we do not allow finishing stream load pipe without data
+ LOG(WARNING) << "Cancelled append msg to pulsar pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel("Cancelled");
+ return Status::InternalError("Cancelled");
+ } else {
+ DCHECK(left_bytes < ctx->max_batch_size);
+ static_cast<void>(pulsar_pipe->finish());
+ ctx->pulsar_info->ack_offset = std::move(ack_offset);
+ ctx->receive_bytes = ctx->max_batch_size - left_bytes;
+ get_backlog_nums(ctx);
+ acknowledge_cumulative(ctx);
+ return Status::OK();
+ }
+ }
+
+ pulsar::Message* msg;
+ bool res = _queue.blocking_get(&msg);
+ if (res) {
+ std::string partition = msg->getProperty("topicName");
+ pulsar::MessageId msg_id = msg->getMessageId();
+ std::size_t len = msg->getLength();
+
+ VLOG(3) << "get pulsar message:" << msg->getDataAsString()
+ << ", partition: " << partition << ", message id: " << msg_id
+ << ", len: " << len << ", size: " << msg->getDataAsString().size();
+
+ Status st = (pulsar_pipe.get()->*append_data)(msg->getDataAsString().c_str(),
+ msg->getDataAsString().size());
+ if (st.ok()) {
+ received_rows++;
+ // len of receive origin message from pulsar
+ left_bytes -= len;
+ if (ack_offset.find(partition) != ack_offset.end()) {
+ if (msg->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+ LOG(INFO) << "old message id: " << ack_offset[partition]
+ << ", new message id: " << msg_id;
+ }
+ if (ack_offset[partition] < msg_id) {
+ ack_offset[partition] = msg_id;
+ }
+ } else {
+ ack_offset[partition] = msg_id;
+ }
+ VLOG(3) << "consume partition" << partition << " - " << msg_id;
+ } else {
+ // failed to append this msg, we must stop
+ LOG(WARNING) << "failed to append msg to pipe. grp: " << _grp_id << ", errmsg=" << st.to_string();
+ eos = true;
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ if (result_st.ok()) {
+ result_st = st;
+ }
+ }
+ }
+ delete msg;
+ } else {
+ // queue is empty and shutdown
+ eos = true;
+ }
+
+ left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
Review Comment:
warning: 1000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
^
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
+ int64_t received_rows = 0;
+ int64_t left_bytes = ctx->max_batch_size;
+
+ std::shared_ptr<io::PulsarConsumerPipe> pulsar_pipe = std::static_pointer_cast<io::PulsarConsumerPipe>(ctx->body_sink);
+
+ LOG(INFO) << "start consumer group: " << _grp_id << ". max time(ms): " << left_time
+ << ", batch size: " << left_bytes << ". " << ctx->brief();
+
+ // copy one
+ std::map<std::string, pulsar::MessageId> ack_offset = ctx->pulsar_info->ack_offset;
+
+ //improve performance
+ Status (io::PulsarConsumerPipe::*append_data)(const char* data, size_t size);
+ if (ctx->format == TFileFormatType::FORMAT_JSON) {
+ append_data = &io::PulsarConsumerPipe::append_json;
+ } else {
+ append_data = &io::PulsarConsumerPipe::append_with_line_delimiter;
+ }
+
+ MonotonicStopWatch watch;
+ watch.start();
+ bool eos = false;
+ while (true) {
+ if (eos || left_time <= 0 || left_bytes <= 0) {
+ LOG(INFO) << "consumer group done: " << _grp_id
+ << ". consume time(ms)=" << ctx->max_interval_s * 1000 - left_time
+ << ", received rows=" << received_rows << ", received bytes=" << ctx->max_batch_size - left_bytes
+ << ", eos: " << eos << ", left_time: " << left_time << ", left_bytes: " << left_bytes
+ << ", blocking get time(us): " << _queue.total_get_wait_time() / 1000
+ << ", blocking put time(us): " << _queue.total_put_wait_time() / 1000;
+
+ // cancel all consumers
+ for (auto& consumer : _consumers) {
+ static_cast<void>(consumer->cancel(ctx));
+ }
+
+ // sleep 3s for waitting consumer cancel done
+ std::this_thread::sleep_for(std::chrono::seconds(3));
+ // shutdown queue
+ _queue.shutdown();
+
+ // waiting all threads finished
+ _thread_pool.shutdown();
+ _thread_pool.join();
+
+ if (!result_st.ok()) {
+ // some consumers encounter errors, cancel this task
+ LOG(WARNING) << "Failed. Cancelled append msg to pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel(result_st.to_string());
+ return result_st;
+ }
+
+ if (left_bytes == ctx->max_batch_size) {
+ // nothing to be consumed, we have to cancel it, because
+ // we do not allow finishing stream load pipe without data
+ LOG(WARNING) << "Cancelled append msg to pulsar pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel("Cancelled");
+ return Status::InternalError("Cancelled");
+ } else {
+ DCHECK(left_bytes < ctx->max_batch_size);
+ static_cast<void>(pulsar_pipe->finish());
+ ctx->pulsar_info->ack_offset = std::move(ack_offset);
+ ctx->receive_bytes = ctx->max_batch_size - left_bytes;
+ get_backlog_nums(ctx);
+ acknowledge_cumulative(ctx);
+ return Status::OK();
+ }
+ }
+
+ pulsar::Message* msg;
+ bool res = _queue.blocking_get(&msg);
+ if (res) {
+ std::string partition = msg->getProperty("topicName");
+ pulsar::MessageId msg_id = msg->getMessageId();
+ std::size_t len = msg->getLength();
+
+ VLOG(3) << "get pulsar message:" << msg->getDataAsString()
+ << ", partition: " << partition << ", message id: " << msg_id
+ << ", len: " << len << ", size: " << msg->getDataAsString().size();
+
+ Status st = (pulsar_pipe.get()->*append_data)(msg->getDataAsString().c_str(),
+ msg->getDataAsString().size());
+ if (st.ok()) {
+ received_rows++;
+ // len of receive origin message from pulsar
+ left_bytes -= len;
+ if (ack_offset.find(partition) != ack_offset.end()) {
+ if (msg->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+ LOG(INFO) << "old message id: " << ack_offset[partition]
+ << ", new message id: " << msg_id;
+ }
+ if (ack_offset[partition] < msg_id) {
+ ack_offset[partition] = msg_id;
+ }
+ } else {
+ ack_offset[partition] = msg_id;
+ }
+ VLOG(3) << "consume partition" << partition << " - " << msg_id;
+ } else {
+ // failed to append this msg, we must stop
+ LOG(WARNING) << "failed to append msg to pipe. grp: " << _grp_id << ", errmsg=" << st.to_string();
+ eos = true;
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ if (result_st.ok()) {
+ result_st = st;
+ }
+ }
+ }
+ delete msg;
+ } else {
+ // queue is empty and shutdown
+ eos = true;
+ }
+
+ left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
Review Comment:
warning: 1000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
^
```
##########
be/src/runtime/routine_load/data_consumer_group.cpp:
##########
@@ -202,4 +205,251 @@
cb(st);
}
+Status PulsarDataConsumerGroup::assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx) {
+ DCHECK(ctx->pulsar_info);
+ DCHECK(_consumers.size() >= 1);
+ // Cumulative acknowledgement when consuming partitioned topics is not supported by pulsar
+ DCHECK(_consumers.size() == ctx->pulsar_info->partitions.size());
+
+ // assign partition to consumers
+ int consumer_size = _consumers.size();
+ for (int i = 0; i < consumer_size; ++i) {
+ auto iter = ctx->pulsar_info->initial_positions.find(ctx->pulsar_info->partitions[i]);
+ if (iter != ctx->pulsar_info->initial_positions.end()) {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx, iter->second));
+ } else {
+ RETURN_IF_ERROR(std::static_pointer_cast<PulsarDataConsumer>(_consumers[i])
+ ->assign_partition(ctx->pulsar_info->partitions[i], ctx));
+ }
+ }
+
+ return Status::OK();
+}
+
+PulsarDataConsumerGroup::~PulsarDataConsumerGroup() {
+ // clean the msgs left in queue
+ _queue.shutdown();
+ while (true) {
+ pulsar::Message* msg;
+ if (_queue.blocking_get(&msg)) {
+ delete msg;
+ msg = nullptr;
+ } else {
+ break;
+ }
+ }
+ DCHECK(_queue.get_size() == 0);
+}
+
+Status PulsarDataConsumerGroup::start_all(std::shared_ptr<StreamLoadContext> ctx) {
+ Status result_st = Status::OK();
+ // start all consumers
+ for (auto& consumer : _consumers) {
+ if (!_thread_pool.offer([this, consumer, capture0 = &_queue, capture1 = ctx->max_interval_s * 1000,
+ capture2 = [this, &result_st](const Status& st) {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _counter--;
+ VLOG(1) << "group counter is: " << _counter << ", grp: " << _grp_id;
+ if (_counter == 0) {
+ _queue.shutdown();
+ LOG(INFO)
+ << "all consumers are finished. shutdown queue. group id: " << _grp_id;
+ }
+ if (result_st.ok() && !st.ok()) {
+ result_st = st;
+ }
+ }] { actual_consume(consumer, capture0, capture1, capture2); })) {
+ LOG(WARNING) << "failed to submit data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ return Status::InternalError("failed to submit data consumer");
+ } else {
+ VLOG(1) << "submit a data consumer: " << consumer->id() << ", group id: " << _grp_id;
+ }
+ }
+
+ // consuming from queue and put data to stream load pipe
+ int64_t left_time = ctx->max_interval_s * 1000;
+ int64_t received_rows = 0;
+ int64_t left_bytes = ctx->max_batch_size;
+
+ std::shared_ptr<io::PulsarConsumerPipe> pulsar_pipe = std::static_pointer_cast<io::PulsarConsumerPipe>(ctx->body_sink);
+
+ LOG(INFO) << "start consumer group: " << _grp_id << ". max time(ms): " << left_time
+ << ", batch size: " << left_bytes << ". " << ctx->brief();
+
+ // copy one
+ std::map<std::string, pulsar::MessageId> ack_offset = ctx->pulsar_info->ack_offset;
+
+ //improve performance
+ Status (io::PulsarConsumerPipe::*append_data)(const char* data, size_t size);
+ if (ctx->format == TFileFormatType::FORMAT_JSON) {
+ append_data = &io::PulsarConsumerPipe::append_json;
+ } else {
+ append_data = &io::PulsarConsumerPipe::append_with_line_delimiter;
+ }
+
+ MonotonicStopWatch watch;
+ watch.start();
+ bool eos = false;
+ while (true) {
+ if (eos || left_time <= 0 || left_bytes <= 0) {
+ LOG(INFO) << "consumer group done: " << _grp_id
+ << ". consume time(ms)=" << ctx->max_interval_s * 1000 - left_time
+ << ", received rows=" << received_rows << ", received bytes=" << ctx->max_batch_size - left_bytes
+ << ", eos: " << eos << ", left_time: " << left_time << ", left_bytes: " << left_bytes
+ << ", blocking get time(us): " << _queue.total_get_wait_time() / 1000
+ << ", blocking put time(us): " << _queue.total_put_wait_time() / 1000;
+
+ // cancel all consumers
+ for (auto& consumer : _consumers) {
+ static_cast<void>(consumer->cancel(ctx));
+ }
+
+ // sleep 3s for waitting consumer cancel done
+ std::this_thread::sleep_for(std::chrono::seconds(3));
+ // shutdown queue
+ _queue.shutdown();
+
+ // waiting all threads finished
+ _thread_pool.shutdown();
+ _thread_pool.join();
+
+ if (!result_st.ok()) {
+ // some consumers encounter errors, cancel this task
+ LOG(WARNING) << "Failed. Cancelled append msg to pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel(result_st.to_string());
+ return result_st;
+ }
+
+ if (left_bytes == ctx->max_batch_size) {
+ // nothing to be consumed, we have to cancel it, because
+ // we do not allow finishing stream load pipe without data
+ LOG(WARNING) << "Cancelled append msg to pulsar pipe. grp: " << _grp_id;
+ pulsar_pipe->cancel("Cancelled");
+ return Status::InternalError("Cancelled");
+ } else {
+ DCHECK(left_bytes < ctx->max_batch_size);
+ static_cast<void>(pulsar_pipe->finish());
+ ctx->pulsar_info->ack_offset = std::move(ack_offset);
+ ctx->receive_bytes = ctx->max_batch_size - left_bytes;
+ get_backlog_nums(ctx);
+ acknowledge_cumulative(ctx);
+ return Status::OK();
+ }
+ }
+
+ pulsar::Message* msg;
+ bool res = _queue.blocking_get(&msg);
+ if (res) {
+ std::string partition = msg->getProperty("topicName");
+ pulsar::MessageId msg_id = msg->getMessageId();
+ std::size_t len = msg->getLength();
+
+ VLOG(3) << "get pulsar message:" << msg->getDataAsString()
+ << ", partition: " << partition << ", message id: " << msg_id
+ << ", len: " << len << ", size: " << msg->getDataAsString().size();
+
+ Status st = (pulsar_pipe.get()->*append_data)(msg->getDataAsString().c_str(),
+ msg->getDataAsString().size());
+ if (st.ok()) {
+ received_rows++;
+ // len of receive origin message from pulsar
+ left_bytes -= len;
+ if (ack_offset.find(partition) != ack_offset.end()) {
+ if (msg->getDataAsString().find("\"country\":\"PL\"") != std::string::npos) {
+ LOG(INFO) << "old message id: " << ack_offset[partition]
+ << ", new message id: " << msg_id;
+ }
+ if (ack_offset[partition] < msg_id) {
+ ack_offset[partition] = msg_id;
+ }
+ } else {
+ ack_offset[partition] = msg_id;
+ }
+ VLOG(3) << "consume partition" << partition << " - " << msg_id;
+ } else {
+ // failed to append this msg, we must stop
+ LOG(WARNING) << "failed to append msg to pipe. grp: " << _grp_id << ", errmsg=" << st.to_string();
+ eos = true;
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ if (result_st.ok()) {
+ result_st = st;
+ }
+ }
+ }
+ delete msg;
+ } else {
+ // queue is empty and shutdown
+ eos = true;
+ }
+
+ left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
Review Comment:
warning: 1000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
left_time = ctx->max_interval_s * 1000 - watch.elapsed_time() / 1000 / 1000;
^
```
##########
be/src/runtime/routine_load/data_consumer_group.h:
##########
@@ -108,4 +106,34 @@
BlockingQueue<RdKafka::Message*> _queue;
};
+// for pulsar
+class PulsarDataConsumerGroup : public DataConsumerGroup {
+public:
+ PulsarDataConsumerGroup() : DataConsumerGroup(), _queue(3000) {}
+
+ ~PulsarDataConsumerGroup() override;
+
+ Status start_all(std::shared_ptr<StreamLoadContext> ctx) override;
+ // assign topic partitions to all consumers equally
+ Status assign_topic_partitions(std::shared_ptr<StreamLoadContext> ctx);
+
+private:
+ // start a single consumer
+ void actual_consume(const std::shared_ptr<DataConsumer>& consumer, BlockingQueue<pulsar::Message*>* queue,
+ int64_t max_running_time_ms, const ConsumeFinishCallback& cb);
+
+ void get_backlog_nums(std::shared_ptr<StreamLoadContext> ctx);
+
+ const char* filter_invalid_prefix_of_json(const char* data, std::size_t size);
+
+ size_t len_of_actual_data(const char* data);
+
+ // acknowledge pulsar message
+ void acknowledge_cumulative(std::shared_ptr<StreamLoadContext> ctx);
+
+private:
Review Comment:
warning: redundant access specifier has the same accessibility as the previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
<details>
<summary>Additional context</summary>
**be/src/runtime/routine_load/data_consumer_group.h:119:** previously declared here
```cpp
private:
^
```
</details>
##########
be/src/runtime/routine_load/routine_load_task_executor.cpp:
##########
@@ -173,6 +173,93 @@
return st;
}
+Status RoutineLoadTaskExecutor::get_pulsar_partition_meta(const PPulsarMetaProxyRequest& request,
+ std::vector<std::string>* partitions) {
+ DCHECK(request.has_pulsar_info());
+
+ // This context is meaningless, just for unifing the interface
+ std::shared_ptr<StreamLoadContext> ctx = std::make_shared<StreamLoadContext>(_exec_env);
+ ctx->load_type = TLoadType::ROUTINE_LOAD;
+ ctx->load_src_type = TLoadSourceType::PULSAR;
+ ctx->label = "NaN";
+
+ // convert PPulsarInfo to TPulsarLoadInfo
+ TPulsarLoadInfo t_info;
+ t_info.service_url = request.pulsar_info().service_url();
+ t_info.topic = request.pulsar_info().topic();
+ t_info.subscription = request.pulsar_info().subscription();
+ std::map<std::string, std::string> properties;
+ for (int i = 0; i < request.pulsar_info().properties_size(); ++i) {
+ const PStringPair& pair = request.pulsar_info().properties(i);
+ properties.emplace(pair.key(), pair.val());
+ }
+ t_info.__set_properties(properties);
+
+ ctx->pulsar_info = std::make_unique<PulsarLoadInfo>(t_info);
+ ctx->need_rollback = false;
+
+ std::shared_ptr<DataConsumer> consumer;
+ RETURN_IF_ERROR(_data_consumer_pool.get_consumer(ctx, &consumer));
+
+ Status st = std::static_pointer_cast<PulsarDataConsumer>(consumer)->get_topic_partition(partitions);
+ if (st.ok()) {
+ _data_consumer_pool.return_consumer(consumer);
+ }
+ return st;
+}
+
+Status RoutineLoadTaskExecutor::get_pulsar_partition_backlog(const PPulsarBacklogProxyRequest& request,
Review Comment:
warning: method 'get_pulsar_partition_backlog' can be made static [readability-convert-member-functions-to-static]
```suggestion
static Status RoutineLoadTaskExecutor::get_pulsar_partition_backlog(const PPulsarBacklogProxyRequest& request,
```
##########
be/src/runtime/routine_load/routine_load_task_executor.cpp:
##########
@@ -336,21 +436,8 @@
#endif
}
- std::shared_ptr<io::KafkaConsumerPipe> kafka_pipe =
- std::static_pointer_cast<io::KafkaConsumerPipe>(ctx->body_sink);
-
// start to consume, this may block a while
- HANDLE_ERROR(consumer_grp->start_all(ctx, kafka_pipe), "consuming failed");
-
- if (ctx->is_multi_table) {
- // plan the rest of unplanned data
- auto multi_table_pipe = std::static_pointer_cast<io::MultiTablePipe>(ctx->body_sink);
- HANDLE_ERROR(multi_table_pipe->request_and_exec_plans(),
- "multi tables task executes plan error");
- // need memory order
- multi_table_pipe->set_consume_finished();
- HANDLE_ERROR(kafka_pipe->finish(), "finish multi table task failed");
- }
+ HANDLE_ERROR(consumer_grp->start_all(ctx), "consuming failed");
Review Comment:
warning: boolean expression can be simplified by DeMorgan's theorem [readability-simplify-boolean-expr]
```cpp
HANDLE_ERROR(consumer_grp->start_all(ctx), "consuming failed");
^
```
<details>
<summary>Additional context</summary>
**be/src/runtime/routine_load/routine_load_task_executor.cpp:370:** expanded from macro 'HANDLE_ERROR'
```cpp
if (UNLIKELY(!_status_.ok() && !_status_.is<PUBLISH_TIMEOUT>())) { \
^
```
**be/src/common/compiler_util.h:35:** expanded from macro 'UNLIKELY'
```cpp
#define UNLIKELY(expr) __builtin_expect(!!(expr), 0)
^
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Pulsar test 1126 [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27802:
URL: https://github.com/apache/doris/pull/27802#issuecomment-1833183239
#### `sh-checker report`
To get the full details, please check in the [job]("https://github.com/apache/doris/actions/runs/7042866695") output.
<details>
<summary>shellcheck errors</summary>
```
'shellcheck ' returned error 1 finding the following syntactical issues:
----------
In thirdparty/build-thirdparty.sh line 904:
check_if_source_exist $PULSAR_SOURCE
^------------^ SC2248 (style): Prefer double quoting even when variables don't contain special characters.
^------------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
Did you mean:
check_if_source_exist "${PULSAR_SOURCE}"
In thirdparty/build-thirdparty.sh line 906:
cd $TP_SOURCE_DIR/$PULSAR_SOURCE
^------------^ SC2086 (info): Double quote to prevent globbing and word splitting.
^------------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
^------------^ SC2248 (style): Prefer double quoting even when variables don't contain special characters.
^------------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
Did you mean:
cd "${TP_SOURCE_DIR}"/"${PULSAR_SOURCE}"
In thirdparty/build-thirdparty.sh line 908:
$CMAKE_CMD -DCMAKE_LIBRARY_PATH=$TP_INSTALL_DIR/lib -DCMAKE_INCLUDE_PATH=$TP_INSTALL_DIR/include \
^--------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
^-------------^ SC2086 (info): Double quote to prevent globbing and word splitting.
^-------------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
^-------------^ SC2086 (info): Double quote to prevent globbing and word splitting.
^-------------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
Did you mean:
${CMAKE_CMD} -DCMAKE_LIBRARY_PATH="${TP_INSTALL_DIR}"/lib -DCMAKE_INCLUDE_PATH="${TP_INSTALL_DIR}"/include \
In thirdparty/build-thirdparty.sh line 909:
-DPROTOC_PATH=$TP_INSTALL_DIR/bin/protoc -DBUILD_TESTS=OFF -DBUILD_PYTHON_WRAPPER=OFF -DBUILD_DYNAMIC_LIB=OFF .
^-------------^ SC2086 (info): Double quote to prevent globbing and word splitting.
^-------------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
Did you mean:
-DPROTOC_PATH="${TP_INSTALL_DIR}"/bin/protoc -DBUILD_TESTS=OFF -DBUILD_PYTHON_WRAPPER=OFF -DBUILD_DYNAMIC_LIB=OFF .
In thirdparty/build-thirdparty.sh line 910:
${BUILD_SYSTEM} -j$PARALLEL
^-------^ SC2086 (info): Double quote to prevent globbing and word splitting.
^-------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
Did you mean:
${BUILD_SYSTEM} -j"${PARALLEL}"
In thirdparty/build-thirdparty.sh line 912:
cp lib/libpulsar.a $TP_INSTALL_DIR/lib/
^-------------^ SC2086 (info): Double quote to prevent globbing and word splitting.
^-------------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
Did you mean:
cp lib/libpulsar.a "${TP_INSTALL_DIR}"/lib/
In thirdparty/build-thirdparty.sh line 913:
cp -r include/pulsar $TP_INSTALL_DIR/include/
^-------------^ SC2086 (info): Double quote to prevent globbing and word splitting.
^-------------^ SC2250 (style): Prefer putting braces around variable references even when not strictly required.
Did you mean:
cp -r include/pulsar "${TP_INSTALL_DIR}"/include/
For more information:
https://www.shellcheck.net/wiki/SC2086 -- Double quote to prevent globbing ...
https://www.shellcheck.net/wiki/SC2248 -- Prefer double quoting even when v...
https://www.shellcheck.net/wiki/SC2250 -- Prefer putting braces around vari...
----------
You can address the above issues in one of three ways:
1. Manually correct the issue in the offending shell script;
2. Disable specific issues by adding the comment:
# shellcheck disable=NNNN
above the line that contains the issue, where NNNN is the error code;
3. Add '-e NNNN' to the SHELLCHECK_OPTS setting in your .yml action file.
```
</details>
<details>
<summary>shfmt errors</summary>
```
'shfmt ' found no issues.
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Pulsar test 1126 [doris]
Posted by "zshben (via GitHub)" <gi...@apache.org>.
zshben closed pull request #27802: Pulsar test 1126
URL: https://github.com/apache/doris/pull/27802
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org