Merge pull request #1254 from kedixa/dev

kafka bugfix: cannot fetch without group

src/client/WFKafkaClient.cc

@@ -28,6 +28,7 @@
 #define KAFKA_CGROUP_UNINIT		0
 #define KAFKA_CGROUP_DOING		1
 #define KAFKA_CGROUP_DONE		2
+#define KAFKA_CGROUP_NONE		3
 
 #define KAFKA_HEARTBEAT_UNINIT	0
 #define KAFKA_HEARTBEAT_DOING	1
@@ -52,7 +53,7 @@ class KafkaMember
 public:
 	KafkaMember() : ref(1)
 	{
-		cgroup_status = KAFKA_CGROUP_UNINIT;
+		cgroup_status = KAFKA_CGROUP_NONE;
 		heartbeat_status = KAFKA_HEARTBEAT_UNINIT;
 		heartbeat_series = NULL;
 		cgroup_outdated = false;
@@ -1587,6 +1588,7 @@ 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;
 }
 

src/factory/KafkaTaskImpl.cc

@@ -287,6 +287,7 @@ CommMessageIn *__ComplexKafkaTask::message_in()
 {
 	KafkaRequest *req = static_cast<KafkaRequest *>(this->get_message_out());
 	KafkaResponse *resp = this->get_resp();
+	KafkaCgroup *cgroup;
 
 	resp->set_api_type(req->get_api_type());
 	resp->set_api_version(req->get_api_version());
@@ -296,7 +297,9 @@ CommMessageIn *__ComplexKafkaTask::message_in()
 	{
 	case Kafka_FindCoordinator:
 	case Kafka_Heartbeat:
-		resp->set_cgroup(__create_cgroup(req->get_cgroup()));
+		cgroup = req->get_cgroup();
+		if (cgroup->get_group())
+			resp->set_cgroup(__create_cgroup(cgroup));
 		break;
 	default:
 		break;

src/manager/RouteManager.cc

@@ -511,9 +511,9 @@ int RouteManager::get(TransportType type,
 
 		struct RouteParams params = {
 			.transport_type			=	type,
-			.addrinfo 				= 	addrinfo,
+			.addrinfo 				=	addrinfo,
 			.key					=	key,
-			.ssl_ctx 				=	ssl_ctx,
+			.ssl_ctx				=	ssl_ctx,
 			.max_connections		=	(unsigned int)ep_params->max_connections,
 			.connect_timeout		=	ep_params->connect_timeout,
 			.response_timeout		=	ep_params->response_timeout,