Add support for kafka client over SSL.

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.

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请求。

src/client/WFKafkaClient.cc

@@ -52,14 +52,15 @@ using ComplexKafkaTask = WFComplexClientTask<KafkaRequest, KafkaResponse,
 class KafkaMember
 {
 public:
-	KafkaMember() : ref(1)
+	KafkaMember() : scheme("kafka://"), ref(1)
 	{
-		cgroup_status = KAFKA_CGROUP_NONE;
-		heartbeat_status = KAFKA_HEARTBEAT_UNINIT;
-		meta_doing = false;
-		cgroup_outdated = false;
-		client_deinit = false;
-		heartbeat_series = NULL;
+		this->transport_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<std::string> 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<std::string> *hosts,
+	static void kafka_merge_broker_list(const std::string& scheme,
+										std::vector<std::string> *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<std::string> *hosts,
+void KafkaClientTask::kafka_merge_broker_list(const std::string& scheme,
+											  std::vector<std::string> *hosts,
 											  KafkaBrokerMap *dst,
 											  KafkaBrokerList *src)
 {
@@ -445,9 +450,8 @@ void KafkaClientTask::kafka_merge_broker_list(std::vector<std::string> *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<int, int> *state_error;
+	std::pair<int32_t, int32_t> *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<int, int> *)pwork->series_at(i)->get_context();
+		state_error = (std::pair<int32_t, int32_t> *)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<int, int> *state_error = new std::pair<int, int>;
+	auto *state_error = new std::pair<int32_t, int32_t>;
+	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<ComplexKafkaTask *>(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<std::string> 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()

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 <assert.h>
@@ -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);

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,

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;
 }

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; }

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)