database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[enhancement](routineload) add debug conf and set broker.name.ttl = 0 (#23302)

Browse Source 
* set broker.name.ttl = 0

* add debug config for librdkafka
This commit is contained in:
Yongqiang YANG
2023-08-26 10:56:35 +08:00
committed by GitHub
parent db8d18eb40
commit bc020112fc
3 changed files with 6 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
be/src/common/config.cpp
View File

@ -770,6 +770,7 @@ DEFINE_mInt32(max_remote_storage_count, "10");
// and the valid values are: 0.9.0.x, 0.8.x.y.
DEFINE_String(kafka_api_version_request, "true");
DEFINE_String(kafka_broker_version_fallback, "0.10.0");
DEFINE_String(kafka_debug, "disable");
// The number of pool siz of routine load consumer.
// If you meet the error describe in https://github.com/edenhill/librdkafka/issues/3608

1
be/src/common/config.h
View File

@ -826,6 +826,7 @@ DECLARE_mInt32(max_remote_storage_count);
// and the valid values are: 0.9.0.x, 0.8.x.y.
DECLARE_String(kafka_api_version_request);
DECLARE_String(kafka_broker_version_fallback);
DECLARE_mString(kafka_debug);
// The number of pool siz of routine load consumer.
// If you meet the error describe in https://github.com/edenhill/librdkafka/issues/3608

4
be/src/runtime/routine_load/data_consumer.cpp
View File

@ -91,6 +91,10 @@ Status KafkaDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
RETURN_IF_ERROR(set_conf("api.version.request", config::kafka_api_version_request));
RETURN_IF_ERROR(set_conf("api.version.fallback.ms", "0"));
RETURN_IF_ERROR(set_conf("broker.version.fallback", config::kafka_broker_version_fallback));
RETURN_IF_ERROR(set_conf("broker.address.ttl", "0"));
if (config::kafka_debug != "disable") {
RETURN_IF_ERROR(set_conf("debug", config::kafka_debug));
}
for (auto& item : ctx->kafka_info->properties) {
if (starts_with(item.second, "FILE:")) {