diff --git a/docs/en/tutorial-13-kafka_cli.md b/docs/en/tutorial-13-kafka_cli.md index 70a16b26..cc08cda7 100644 --- a/docs/en/tutorial-13-kafka_cli.md +++ b/docs/en/tutorial-13-kafka_cli.md @@ -29,8 +29,9 @@ The program exists automatically after all the tasks are completed, and all the In the command, the broker_url may contain several urls seperated by comma(,). -- For instance, kafka://host:port,kafka://host1:port... -- The default value is 9092; +- For instance, kafka://host:port,kafka://host1:port... or: **kafkas**://host:port,**kafkas**://host1:port for kafka over SSL; +- The default port is 9092 for TCP and 9093 for SSL; +- Do not mix 'kafkas://' with "kafka://", otherwise the init function will fail with errno EINVAL; - If you want to use upstream policy at this layer, please refer to [upstream documents](/docs/en/about-upstream.md). The following are several Kafka broker_url samples: @@ -41,6 +42,12 @@ kafka://kafka.host:9090/ kafka://10.160.23.23:9000,10.123.23.23,kafka://kafka.sogou +kafkas://broker1.kafka.sogou,kafkas://broker2.kafka.sogou + +Illegal broker_url sample (The first one is SSL, and the second one is not): + +kafkas://broker1.kafka.sogou,broker2.kafka.sogou + # Principles and Features Kafka client has no third-party dependencies internally except for the libraries used in the compression. With the high performance of Workflow, When properly configured and in fair environments, tens of thousands of Kafka requests can be processed in one second. diff --git a/docs/tutorial-13-kafka_cli.md b/docs/tutorial-13-kafka_cli.md index a3fcf532..714846e9 100644 --- a/docs/tutorial-13-kafka_cli.md +++ b/docs/tutorial-13-kafka_cli.md @@ -28,8 +28,10 @@ Bazel：执行bazel build kafka 编译支持kafka协议的类库；执行bazel b 其中broker_url可以有多个url组成，多个url之间以,分割 -- 形式如：kafka://host:port,kafka://host1:port... -- port默认为9092; +- 形式如：kafka://host:port,kafka://host1:port... 或：**kafkas**://host:port,**kafkas**://host1:port代表使用SSL通信。 +- port的默认值在普通TCP连接下是9092，SSL下为9093。 +- "kafka://"前缀可以缺省。这时候使用默认使用TCL通信。 +- 多个url，必须都采用TCP或都采用SSL。否则init函数返回-1，错误码为EINVAL。 - 如果用户在这一层有upstream选取需求，可以参考[upstream文档](../docs/about-upstream.md)。 Kafka broker_url示例： @@ -40,6 +42,12 @@ kafka://kafka.host:9090/ kafka://10.160.23.23:9000,10.123.23.23,kafka://kafka.sogou +kafkas://broker1.kafka.sogou,kafkas://broker2.kafka.sogou + +错误的url示例（第一个broker为SSL，第二个broker非SSL）： + +kafkas://broker1.kafka.sogou,broker2.kafka.sogou + # 实现原理和特性 kafka client内部实现上除了压缩功能外没有依赖第三方库，同时利用了workflow的高性能，在合理的配置和环境下，每秒钟可以处理几万次Kafka请求。 diff --git a/src/client/WFKafkaClient.cc b/src/client/WFKafkaClient.cc index 88666e2e..9a84e004 100644 --- a/src/client/WFKafkaClient.cc +++ b/src/client/WFKafkaClient.cc @@ -52,14 +52,15 @@ using ComplexKafkaTask = WFComplexClientTasktransport_type = TT_TCP; + this->cgroup_status = KAFKA_CGROUP_NONE; + this->heartbeat_status = KAFKA_HEARTBEAT_UNINIT; + this->meta_doing = false; + this->cgroup_outdated = false; + this->client_deinit = false; + this->heartbeat_series = NULL; } void incref() @@ -73,6 +74,8 @@ public: delete this; } + TransportType transport_type; + std::string scheme; std::vector broker_hosts; KafkaCgroup cgroup; KafkaMetaList meta_list; @@ -146,7 +149,8 @@ private: static void kafka_merge_meta_list(KafkaMetaList *dst, KafkaMetaList *src); - static void kafka_merge_broker_list(std::vector *hosts, + static void kafka_merge_broker_list(const std::string& scheme, + std::vector *hosts, KafkaBrokerMap *dst, KafkaBrokerList *src); @@ -292,10 +296,8 @@ void KafkaClientTask::kafka_rebalance_callback(__WFKafkaTask *task) socklen_t socklen; coordinator->get_broker_addr(&addr, &socklen); - kafka_task = __WFKafkaTaskFactory::create_kafka_task(addr, - socklen, - "", - 0, + kafka_task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + addr, socklen, "", 0, kafka_heartbeat_callback); kafka_task->user_data = member; kafka_task->get_req()->set_api_type(Kafka_Heartbeat); @@ -327,7 +329,8 @@ void KafkaClientTask::kafka_rebalance_proc(KafkaMember *member, SeriesWork *seri coordinator->get_broker_addr(&addr, &socklen); __WFKafkaTask *task; - task = __WFKafkaTaskFactory::create_kafka_task(addr, socklen, "", 0, + task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + addr, socklen, "", 0, kafka_rebalance_callback); task->user_data = member; task->get_req()->set_config(member->config); @@ -399,7 +402,8 @@ void KafkaClientTask::kafka_timer_callback(WFTimerTask *task) socklen_t socklen; coordinator->get_broker_addr(&addr, &socklen); - kafka_task = __WFKafkaTaskFactory::create_kafka_task(addr, socklen, "", 0, + kafka_task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + addr, socklen, "", 0, kafka_heartbeat_callback); kafka_task->user_data = member; @@ -436,7 +440,8 @@ void KafkaClientTask::kafka_merge_meta_list(KafkaMetaList *dst, } } -void KafkaClientTask::kafka_merge_broker_list(std::vector *hosts, +void KafkaClientTask::kafka_merge_broker_list(const std::string& scheme, + std::vector *hosts, KafkaBrokerMap *dst, KafkaBrokerList *src) { @@ -445,9 +450,8 @@ void KafkaClientTask::kafka_merge_broker_list(std::vector *hosts, KafkaBroker *src_broker; while ((src_broker = src->get_next()) != NULL) { - std::string host = "kafka://"; - host = host + src_broker->get_host() + ":" + - std::to_string(src_broker->get_port()); + std::string host = scheme + src_broker->get_host() + ":" + + std::to_string(src_broker->get_port()); hosts->emplace_back(std::move(host)); if (!dst->find_item(src_broker->get_node_id())) @@ -475,7 +479,8 @@ void KafkaClientTask::kafka_meta_callback(__WFKafkaTask *task) while ((meta = t->meta_list.get_next()) != NULL) (t->member->meta_status)[meta->get_topic()] = true; - kafka_merge_broker_list(&t->member->broker_hosts, + kafka_merge_broker_list(t->member->scheme, + &t->member->broker_hosts, &t->member->broker_map, task->get_resp()->get_broker_list()); } @@ -523,7 +528,8 @@ void KafkaClientTask::kafka_cgroup_callback(__WFKafkaTask *task) while ((meta = t->meta_list.get_next()) != NULL) (t->member->meta_status)[meta->get_topic()] = true; - kafka_merge_broker_list(&t->member->broker_hosts, + kafka_merge_broker_list(t->member->scheme, + &t->member->broker_hosts, &t->member->broker_map, task->get_resp()->get_broker_list()); @@ -538,8 +544,8 @@ void KafkaClientTask::kafka_cgroup_callback(__WFKafkaTask *task) socklen_t socklen; coordinator->get_broker_addr(&addr, &socklen); - kafka_task = __WFKafkaTaskFactory::create_kafka_task(addr, socklen, - "", 0, + kafka_task = __WFKafkaTaskFactory::create_kafka_task(t->member->transport_type, + addr, socklen, "", 0, kafka_heartbeat_callback); kafka_task->user_data = t->member; t->member->incref(); @@ -581,14 +587,14 @@ void KafkaClientTask::kafka_parallel_callback(const ParallelWork *pwork) t->state = WFT_STATE_TASK_ERROR; t->error = 0; - std::pair *state_error; + std::pair *state_error; bool flag = false; - int state = WFT_STATE_SUCCESS; - int error = 0; + int16_t state = WFT_STATE_SUCCESS; + int16_t error = 0; int kafka_error = 0; for (size_t i = 0; i < pwork->size(); i++) { - state_error = (std::pair *)pwork->series_at(i)->get_context(); + state_error = (std::pair *)pwork->series_at(i)->get_context(); if ((state_error->first >> 16) != WFT_STATE_SUCCESS) { if (!flag) @@ -642,9 +648,20 @@ void KafkaClientTask::kafka_process_toppar_offset(KafkaToppar *task_toppar) void KafkaClientTask::kafka_move_task_callback(__WFKafkaTask *task) { - std::pair *state_error = new std::pair; + auto *state_error = new std::pair; + int16_t state = task->get_state(); + int16_t error = task->get_error(); - state_error->first = (task->get_state() << 16) + task->get_error(); + /* This function is called before WFClientTask::done. Need to transfer + the state and error. */ + if (state == WFT_STATE_SYS_ERROR && error < 0) + { + state = WFT_STATE_SSL_ERROR; + error = -error; + } + + /* 'state' is always positive. */ + state_error->first = (state << 16) | error; state_error->second = static_cast(task)->get_mutable_ctx()->kafka_error; series_of(task)->set_context(state_error); @@ -674,27 +691,29 @@ void KafkaClientTask::generate_info() if (this->config.get_sasl_mech()) { - this->userinfo = this->config.get_sasl_username(); + const char *username = this->config.get_sasl_username(); + const char *password = this->config.get_sasl_password(); + + this->userinfo.clear(); + if (username) + this->userinfo += StringUtil::url_encode_component(username); this->userinfo += ":"; - this->userinfo += - StringUtil::url_encode_component(this->config.get_sasl_password()); + if (password) + this->userinfo += StringUtil::url_encode_component(password); this->userinfo += ":"; this->userinfo += this->config.get_sasl_mech(); this->userinfo += ":"; this->userinfo += std::to_string((intptr_t)this->member); - this->userinfo += "@"; - this->url = "kafka://" + this->userinfo + - this->url.substr(this->url.find("kafka://") + 8); } else { char buf[64]; - snprintf(buf, sizeof(buf), "user:pass:sasl:%p@", this->member); + snprintf(buf, 64, "user:pass:sasl:%p", this->member); this->userinfo = buf; - this->url = "kafka://" + this->userinfo + - this->url.substr(this->url.find("kafka://") + 8); } + const char *hostport = this->url.c_str() + this->member->scheme.size(); + this->url = this->member->scheme + this->userinfo + "@" + hostport; this->info_generated = true; } @@ -876,15 +895,16 @@ bool KafkaClientTask::check_meta() int KafkaClientTask::dispatch_locked() { + KafkaMember *member = this->member; __WFKafkaTask *task; ParallelWork *parallel; SeriesWork *series; if (this->check_cgroup() == false) - return this->member->cgroup_wait_cnt > 0; + return member->cgroup_wait_cnt > 0; if (this->check_meta() == false) - return this->member->meta_wait_cnt > 0; + return member->meta_wait_cnt > 0; if (arrange_toppar(this->api_type) < 0) { @@ -925,14 +945,16 @@ int KafkaClientTask::dispatch_locked() socklen_t socklen; broker->get_broker_addr(&addr, &socklen); - task = __WFKafkaTaskFactory::create_kafka_task(addr, socklen, + task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + addr, socklen, this->get_userinfo(), this->retry_max, nullptr); } else { - task = __WFKafkaTaskFactory::create_kafka_task(broker->get_host(), + task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + broker->get_host(), broker->get_port(), this->get_userinfo(), this->retry_max, @@ -976,14 +998,16 @@ int KafkaClientTask::dispatch_locked() socklen_t socklen; broker->get_broker_addr(&addr, &socklen); - task = __WFKafkaTaskFactory::create_kafka_task(addr, socklen, + task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + addr, socklen, this->get_userinfo(), this->retry_max, nullptr); } else { - task = __WFKafkaTaskFactory::create_kafka_task(broker->get_host(), + task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + broker->get_host(), broker->get_port(), this->get_userinfo(), this->retry_max, @@ -1010,7 +1034,7 @@ int KafkaClientTask::dispatch_locked() break; case Kafka_OffsetCommit: - if (!this->member->cgroup.get_group()) + if (!member->cgroup.get_group()) { this->state = WFT_STATE_TASK_ERROR; this->error = WFT_ERR_KAFKA_COMMIT_FAILED; @@ -1020,19 +1044,20 @@ int KafkaClientTask::dispatch_locked() else { this->result.create(1); - KafkaBroker *coordinator = this->member->cgroup.get_coordinator(); + KafkaBroker *coordinator = member->cgroup.get_coordinator(); const struct sockaddr *addr; socklen_t socklen; coordinator->get_broker_addr(&addr, &socklen); - task = __WFKafkaTaskFactory::create_kafka_task(addr, socklen, + task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + addr, socklen, this->get_userinfo(), this->retry_max, kafka_offsetcommit_callback); task->user_data = this; task->get_req()->set_config(this->config); - task->get_req()->set_cgroup(this->member->cgroup); + task->get_req()->set_cgroup(member->cgroup); task->get_req()->set_broker(*coordinator); task->get_req()->set_toppar_list(this->toppar_list); task->get_req()->set_api_type(this->api_type); @@ -1042,7 +1067,7 @@ int KafkaClientTask::dispatch_locked() } case Kafka_LeaveGroup: - if (!this->member->cgroup.get_group()) + if (!member->cgroup.get_group()) { this->state = WFT_STATE_TASK_ERROR; this->error = WFT_ERR_KAFKA_LEAVEGROUP_FAILED; @@ -1051,7 +1076,7 @@ int KafkaClientTask::dispatch_locked() } else { - KafkaBroker *coordinator = this->member->cgroup.get_coordinator(); + KafkaBroker *coordinator = member->cgroup.get_coordinator(); const struct sockaddr *addr; socklen_t socklen; @@ -1059,16 +1084,15 @@ int KafkaClientTask::dispatch_locked() if (coordinator->is_to_addr()) { - task = __WFKafkaTaskFactory::create_kafka_task(addr, - socklen, - this->get_userinfo(), - 0, + task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + addr, socklen, + this->get_userinfo(), 0, kafka_leavegroup_callback); task->user_data = this; task->get_req()->set_config(this->config); task->get_req()->set_api_type(Kafka_LeaveGroup); task->get_req()->set_broker(*coordinator); - task->get_req()->set_cgroup(this->member->cgroup); + task->get_req()->set_cgroup(member->cgroup); series_of(this)->push_front(this); series_of(this)->push_front(task); } @@ -1098,14 +1122,16 @@ int KafkaClientTask::dispatch_locked() socklen_t socklen; broker->get_broker_addr(&addr, &socklen); - task = __WFKafkaTaskFactory::create_kafka_task(addr, socklen, + task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + addr, socklen, this->get_userinfo(), this->retry_max, nullptr); } else { - task = __WFKafkaTaskFactory::create_kafka_task(broker->get_host(), + task = __WFKafkaTaskFactory::create_kafka_task(member->transport_type, + broker->get_host(), broker->get_port(), this->get_userinfo(), this->retry_max, @@ -1635,46 +1661,62 @@ SubTask *WFKafkaTask::done() int WFKafkaClient::init(const std::string& broker) { std::vector broker_hosts; - std::string::size_type pos = broker.find(','); std::string::size_type ppos = 0; + std::string::size_type pos; + bool use_ssl; - if (pos == std::string::npos) + use_ssl = (strncasecmp(broker.c_str(), "kafkas://", 9) == 0); + while (1) { - std::string host = broker; - if (strncasecmp(host.c_str(), "kafka://", 8) != 0) - host = "kafka://" + host; - broker_hosts.emplace_back(host); - } - else - { - do - { - std::string host = broker.substr(ppos, pos - ppos); - if (strncasecmp(host.c_str(), "kafka://", 8) != 0) - host = "kafka://" + host; - broker_hosts.emplace_back(host); - - ppos = pos + 1; - pos = broker.find(',', ppos); - } while (pos != std::string::npos); - + pos = broker.find(',', ppos); std::string host = broker.substr(ppos, pos - ppos); - if (strncasecmp(host.c_str(), "kafka://", 8) != 0) + if (use_ssl) + { + if (strncasecmp(host.c_str(), "kafkas://", 9) != 0) + { + errno = EINVAL; + return -1; + } + } + else if (strncasecmp(host.c_str(), "kafka://", 8) != 0) + { + if (strncasecmp(host.c_str(), "kafkas://", 9) == 0) + { + errno = EINVAL; + return -1; + } + host = "kafka://" + host; + } + broker_hosts.emplace_back(host); + if (pos == std::string::npos) + break; + + ppos = pos + 1; } this->member = new KafkaMember; this->member->broker_hosts = std::move(broker_hosts); + if (use_ssl) + { + this->member->transport_type = TT_TCP_SSL; + this->member->scheme = "kafkas://"; + } + return 0; } int WFKafkaClient::init(const std::string& broker, const std::string& group) { - this->init(broker); - this->member->cgroup.set_group(group); - this->member->cgroup_status = KAFKA_CGROUP_UNINIT; - return 0; + if (this->init(broker) == 0) + { + this->member->cgroup.set_group(group); + this->member->cgroup_status = KAFKA_CGROUP_UNINIT; + return 0; + } + else + return -1; } int WFKafkaClient::deinit() diff --git a/src/factory/KafkaTaskImpl.cc b/src/factory/KafkaTaskImpl.cc index 4bc69196..dcddbc05 100644 --- a/src/factory/KafkaTaskImpl.cc +++ b/src/factory/KafkaTaskImpl.cc @@ -14,6 +14,7 @@ limitations under the License. Authors: Wang Zhulei (wangzhulei@sogou-inc.com) + Xie Han (xiehan@sogou-inc.com) */ #include @@ -310,19 +311,17 @@ CommMessageIn *__ComplexKafkaTask::message_in() bool __ComplexKafkaTask::init_success() { - TransportType type = TT_TCP; - if (uri_.scheme) + TransportType type; + + if (uri_.scheme && strcasecmp(uri_.scheme, "kafka") == 0) + type = TT_TCP; + else if (uri_.scheme && strcasecmp(uri_.scheme, "kafkas") == 0) + type = TT_TCP_SSL; + else { - if (strcasecmp(uri_.scheme, "kafka") == 0) - type = TT_TCP; - //else if (uri_.scheme && strcasecmp(uri_.scheme, "kafkas") == 0) - // type = TT_TCP_SSL; - else - { - this->state = WFT_STATE_TASK_ERROR; - this->error = WFT_ERR_URI_SCHEME_INVALID; - return false; - } + this->state = WFT_STATE_TASK_ERROR; + this->error = WFT_ERR_URI_SCHEME_INVALID; + return false; } std::string username, password, sasl, client; @@ -367,7 +366,6 @@ bool __ComplexKafkaTask::init_success() } this->WFComplexClientTask::set_transport_type(type); - return true; } @@ -426,12 +424,13 @@ bool __ComplexKafkaTask::check_redirect() socklen_t addrlen_coord; coordinator->get_broker_addr(&addr_coord, &addrlen_coord); - set_redirect(TT_TCP, addr_coord, addrlen_coord, + set_redirect(this->get_transport_type(), addr_coord, addrlen_coord, this->WFComplexClientTask::info_); } else { - std::string url = "kafka://"; + std::string url(uri_.scheme); + url += "://"; url += user_info_ + "@"; url += coordinator->get_host(); url += ":" + std::to_string(coordinator->get_port()); @@ -445,7 +444,8 @@ bool __ComplexKafkaTask::check_redirect() } else { - this->init(TT_TCP, paddr, addrlen, this->WFComplexClientTask::info_); + this->init(this->get_transport_type(), paddr, addrlen, + this->WFComplexClientTask::info_); return false; } } @@ -809,7 +809,8 @@ __WFKafkaTask *__WFKafkaTaskFactory::create_kafka_task(const ParsedURI& uri, return task; } -__WFKafkaTask *__WFKafkaTaskFactory::create_kafka_task(const struct sockaddr *addr, +__WFKafkaTask *__WFKafkaTaskFactory::create_kafka_task(TransportType type, + const struct sockaddr *addr, socklen_t addrlen, const std::string& info, int retry_max, @@ -817,12 +818,13 @@ __WFKafkaTask *__WFKafkaTaskFactory::create_kafka_task(const struct sockaddr *ad { auto *task = new __ComplexKafkaTask(retry_max, std::move(callback)); - task->init(TT_TCP, addr, addrlen, info); + task->init(type, addr, addrlen, info); task->set_keep_alive(KAFKA_KEEPALIVE_DEFAULT); return task; } -__WFKafkaTask *__WFKafkaTaskFactory::create_kafka_task(const char *host, +__WFKafkaTask *__WFKafkaTaskFactory::create_kafka_task(TransportType type, + const char *host, unsigned short port, const std::string& info, int retry_max, @@ -830,10 +832,10 @@ __WFKafkaTask *__WFKafkaTaskFactory::create_kafka_task(const char *host, { auto *task = new __ComplexKafkaTask(retry_max, std::move(callback)); - std::string url = "kafka://"; + std::string url = (type == TT_TCP_SSL ? "kafkas://" : "kafka://"); if (!info.empty()) - url += info; + url += info + "@"; url += host; url += ":" + std::to_string(port); diff --git a/src/factory/KafkaTaskImpl.inl b/src/factory/KafkaTaskImpl.inl index cb18c9a5..fd40f4a3 100644 --- a/src/factory/KafkaTaskImpl.inl +++ b/src/factory/KafkaTaskImpl.inl @@ -40,13 +40,15 @@ public: int retry_max, __kafka_callback_t callback); - static __WFKafkaTask *create_kafka_task(const struct sockaddr *addr, + static __WFKafkaTask *create_kafka_task(TransportType type, + const struct sockaddr *addr, socklen_t addrlen, const std::string& info, int retry_max, __kafka_callback_t callback); - static __WFKafkaTask *create_kafka_task(const char *host, + static __WFKafkaTask *create_kafka_task(TransportType type, + const char *host, unsigned short port, const std::string& info, int retry_max, diff --git a/src/manager/WFGlobal.cc b/src/manager/WFGlobal.cc index 0d9b2d43..8d4e20ff 100644 --- a/src/manager/WFGlobal.cc +++ b/src/manager/WFGlobal.cc @@ -154,6 +154,11 @@ __WFGlobal::__WFGlobal() static_scheme_port_["Kafka"] = "9092"; static_scheme_port_["KAFKA"] = "9092"; + static_scheme_port_["kafkas"] = "9093"; + static_scheme_port_["Kafkas"] = "9093"; + static_scheme_port_["KAFKAs"] = "9093"; + static_scheme_port_["KAFKAS"] = "9093"; + sync_count_ = 0; sync_max_ = 0; } diff --git a/src/protocol/KafkaDataTypes.h b/src/protocol/KafkaDataTypes.h index 2d3d546d..1912574d 100644 --- a/src/protocol/KafkaDataTypes.h +++ b/src/protocol/KafkaDataTypes.h @@ -1016,14 +1016,12 @@ public: return this->ptr->port; } - std::string get_uri() const + std::string get_host_port() const { - std::string uri = "kafka://"; - - uri += this->ptr->host; - uri += ":"; - uri += std::to_string(this->ptr->port); - return uri; + std::string host_port(this->ptr->host); + host_port += ":"; + host_port += std::to_string(this->ptr->port); + return host_port; } int get_error() @@ -1103,12 +1101,12 @@ public: bool operator< (const KafkaBroker& broker) const { - return this->get_uri() < broker.get_uri(); + return this->get_host_port() < broker.get_host_port(); } bool operator> (const KafkaBroker& broker) const { - return this->get_uri() > broker.get_uri(); + return this->get_host_port() > broker.get_host_port(); } kafka_broker_t *get_raw_ptr() const { return this->ptr; } diff --git a/src/protocol/kafka_parser.c b/src/protocol/kafka_parser.c index 26e735b6..4f852437 100644 --- a/src/protocol/kafka_parser.c +++ b/src/protocol/kafka_parser.c @@ -1209,6 +1209,8 @@ void kafka_sasl_init(kafka_sasl_t *sasl) sasl->scram.first_msg.iov_len = 0; sasl->scram.server_signature_b64.iov_base = NULL; sasl->scram.server_signature_b64.iov_len = 0; + sasl->buf = NULL; + sasl->bsize = 0; sasl->status = 0; } @@ -1216,6 +1218,7 @@ void kafka_sasl_deinit(kafka_sasl_t *sasl) { free(sasl->scram.cnonce.iov_base); free(sasl->scram.server_signature_b64.iov_base); + free(sasl->buf); } int kafka_sasl_set_username(const char *username, kafka_config_t *conf)