set kafka correlation id

src/client/WFKafkaClient.cc

@@ -521,6 +521,7 @@ void ComplexKafkaTask::kafka_parallel_callback(const ParallelWork *pwork)
 
 	std::pair<int, int> *state_error;
 	bool flag = false;
+	int state = WFT_STATE_SUCCESS;
 	int error = 0;
 	for (size_t i = 0; i < pwork->size(); i++)
 	{
@@ -534,6 +535,7 @@ void ComplexKafkaTask::kafka_parallel_callback(const ParallelWork *pwork)
 				t->set_meta_status(META_UNINIT);
 				t->member->mutex.unlock();
 			}
+			state = state_error->first;
 			error = state_error->second;
 		}
 		else
@@ -545,7 +547,10 @@ void ComplexKafkaTask::kafka_parallel_callback(const ParallelWork *pwork)
 	}
 
 	if (t->state != WFT_STATE_SUCCESS)
+	{
+		t->state = state;
 		t->error = error;
+	}
 }
 
 void ComplexKafkaTask::kafka_process_toppar_offset(KafkaToppar *task_toppar)
@@ -794,6 +799,7 @@ void ComplexKafkaTask::dispatch()
 			this->finish = true;
 			break;
 		}
+
 		parallel = Workflow::create_parallel_work(kafka_parallel_callback);
 		this->result.create(this->toppar_list_map.size());
 		parallel->set_context(this);
@@ -966,6 +972,7 @@ void ComplexKafkaTask::dispatch()
 			this->finish = true;
 			break;
 		}
+
 		parallel = Workflow::create_parallel_work(kafka_parallel_callback);
 		this->result.create(this->toppar_list_map.size());
 		parallel->set_context(this);

src/factory/KafkaTaskImpl.cc

@@ -182,6 +182,7 @@ CommMessageOut *__ComplexKafkaTask::message_out()
 				req->set_api_type(Kafka_SaslHandshake);
 			else
 				req->set_api_type(Kafka_SaslAuthenticate);
+			req->set_correlation_id(1);
 			is_user_request_ = false;
 			return req;
 		}
@@ -256,10 +257,12 @@ CommMessageOut *__ComplexKafkaTask::message_out()
 			new_req->set_config(*req->get_config());
 			new_req->set_api_type(Kafka_ListOffsets);
 			is_user_request_ = false;
+			new_req->set_correlation_id(seqid);
 			return new_req;
 		}
 	}
 
+	this->get_req()->set_correlation_id(seqid);
 	return this->WFComplexClientTask::message_out();
 }
 

src/protocol/KafkaMessage.cc

@@ -1656,6 +1656,7 @@ KafkaMessage::KafkaMessage()
 	kafka_parser_init(this->parser);
 	this->stream = new EncodeStream;
 	this->api_type = Kafka_Unknown;
+	this->correlation_id = 0;
 	this->cur_size = 0;
 }
 
@@ -1799,7 +1800,7 @@ int KafkaMessage::encode_head()
 	append_i32(this->headbuf, 0);
 	append_i16(this->headbuf, this->api_type);
 	append_i16(this->headbuf, this->api_version);
-	append_i32(this->headbuf, 0);
+	append_i32(this->headbuf, this->correlation_id);
 	append_string(this->headbuf, this->config.get_client_id());
 
 	return 0;

src/protocol/KafkaMessage.h

@@ -60,6 +60,9 @@ public:
 	void set_api_version(int ver) { this->api_version = ver; }
 	int get_api_version() const { return this->api_version; }
 
+	void set_correlation_id(int id) { this->correlation_id = id; }
+	int get_correlation_id() const { return this->correlation_id; }
+
 	void set_config(const KafkaConfig& conf)
 	{
 		this->config = conf;
@@ -197,6 +200,7 @@ protected:
 
 	int api_type;
 	int api_version;
+	int correlation_id;
 	int message_version;
 
 	std::map<int, int> api_mver_map;