Merge pull request #155 from wzl12356/master

fix fetch and cgroup bug

docs/tutorial-13-kafka_cli.md

@@ -5,14 +5,21 @@
 
 # 关于编译选项
 
-通过命令make KAFKA=y 编译支持kafka协议的workflow，系统需要预先安装zlib,snappy,lz4(>=1.7.5),zstd等第三方库。
+在workflow中，你可以使用第三方库比如librdkafka，也可使用自带的kafka client，因此它对kafka协议的支持是独立的。
+
+通过命令make KAFKA=y 编译独立的类库支持kafka协议，系统需要预先安装[zlib](https://github.com/madler/zlib.git),[snappy](https://github.com/google/snappy.git),[lz4(>=1.7.5)](https://github.com/lz4/lz4.git),[zstd](https://github.com/facebook/zstd.git)等第三方库。
 
 # 关于kafka_cli
 
-程序从命令行读取一个kafka broker服务器地址和本次任务的类型(produce/fetch/meta)
+这是一个kafka client，根据不同的输入参数，完成kafka的消息生产(produce)、消息消费(fetch)、元数据获取(meta)等。
+
+编译时需要在tutorial目录中执行编译命令make KAFKA=y。
+
+该程序从命令行读取一个kafka broker服务器地址和本次任务的类型(produce/fetch/meta)：
+
 ./kafka_cli \<broker_url\> [p/c/m]
 
-程序会在执行完任务后自动退出
+程序会在执行完任务后自动退出，一切资源完全回收。
 
 其中broker_url可以有多个url组成，多个url之间以,分割
 

src/client/WFKafkaClient.cc

@@ -483,7 +483,7 @@ void ComplexKafkaTask::kafka_meta_callback(__WFKafkaTask *task)
 {
 	ComplexKafkaTask *t = (ComplexKafkaTask *)task->user_data;
 	t->lock_status.get_mutex()->lock();
-	if (task->get_state() == 0)
+	if (task->get_state() == WFT_STATE_SUCCESS)
 	{
 		kafka_merge_meta_list(&t->client_meta_list,
 							  task->get_resp()->get_meta_list());
@@ -493,7 +493,8 @@ void ComplexKafkaTask::kafka_meta_callback(__WFKafkaTask *task)
 		*t->lock_status.get_status() |= KAFKA_META_DONE;
 		*t->lock_status.get_status() &= (~(KAFKA_META_INIT|KAFKA_META_DOING));
 
-		t->state = 0;
+		t->state = WFT_STATE_SUCCESS;
+		t->error = 0;
 	}
 	else
 	{
@@ -547,7 +548,8 @@ void ComplexKafkaTask::kafka_cgroup_callback(__WFKafkaTask *task)
 			*t->lock_status.get_status() &= ~KAFKA_HEARTBEAT_INIT;
 		}
 
-		t->state = 0;
+		t->state = WFT_STATE_SUCCESS;
+		t->error = 0;
 	}
 	else
 	{
@@ -569,15 +571,32 @@ void ComplexKafkaTask::kafka_parallel_callback(const ParallelWork *pwork)
 {
 	ComplexKafkaTask *t = (ComplexKafkaTask *)pwork->get_context();
 	t->finish = true;
-	t->state = 0;
+	t->state = WFT_STATE_SUCCESS;
+	t->error = 0;
+
+	std::pair<int, int> *state_error;
+
+	for (size_t i = 0; i < pwork->size(); i++)
+	{
+		state_error = (std::pair<int, int> *)pwork->series_at(i)->get_context();
+		if (state_error->first != WFT_STATE_SUCCESS)
+		{
+			t->state = state_error->first;
+			t->error = state_error->second;
+		}
+
+		delete state_error;
+	}
 }
 
 void ComplexKafkaTask::kafka_process_toppar_offset(KafkaToppar *task_toppar)
 {
 	KafkaToppar *toppar;
 
-	while ((toppar = this->cgroup.get_assigned_toppar_next()) != NULL)
+	struct list_head *pos;
+	list_for_each(pos, this->cgroup.get_assigned_toppar_list())
 	{
+		toppar = this->cgroup.get_assigned_toppar_by_pos(pos);
 		if (strcmp(toppar->get_topic(), task_toppar->get_topic()) == 0 &&
 			toppar->get_partition() == task_toppar->get_partition())
 		{
@@ -603,6 +622,12 @@ void ComplexKafkaTask::kafka_process_toppar_offset(KafkaToppar *task_toppar)
 
 void ComplexKafkaTask::kafka_move_task_callback(__WFKafkaTask *task)
 {
+	std::pair<int, int> *state_error = new std::pair<int, int>;
+
+	state_error->first = task->get_state();
+	state_error->second = task->get_error();
+	series_of(task)->set_context(state_error);
+
 	KafkaTopparList *toppar_list = task->get_resp()->get_toppar_list();
 
 	if (task->get_resp()->get_api() == Kafka_Fetch)
@@ -611,10 +636,7 @@ void ComplexKafkaTask::kafka_move_task_callback(__WFKafkaTask *task)
 		KafkaToppar *task_toppar;
 
 		while ((task_toppar = toppar_list->get_next()) != NULL)
-		{
-			this->cgroup.assigned_toppar_rewind();
 			kafka_process_toppar_offset(task_toppar);
-		}
 	}
 
 	long idx = (long)(task->user_data);
@@ -698,7 +720,7 @@ void ComplexKafkaTask::dispatch()
 	}
 
 	if ((this->api_type == Kafka_Fetch || this->api_type == Kafka_OffsetCommit) &&
-        (*this->lock_status.get_status() & KAFKA_CGROUP_INIT))
+		(*this->lock_status.get_status() & KAFKA_CGROUP_INIT))
 	{
 		task = __WFKafkaTaskFactory::create_kafka_task(this->uri,
 													   this->retry_max,

src/factory/KafkaTaskImpl.cc

@@ -402,6 +402,17 @@ bool __ComplexKafkaTask::finish_once()
 				(*get_mutable_ctx())(this);
 		}
 	}
+	else
+	{
+		this->disable_retry();
+
+		this->get_resp()->set_api(this->get_req()->get_api());
+		this->get_resp()->set_api_version(this->get_req()->get_api_version());
+		this->get_resp()->duplicate(*this->get_req());
+
+		if (*get_mutable_ctx())
+			(*get_mutable_ctx())(this);
+	}
 
 	return true;
 }

src/protocol/KafkaDataTypes.cc

@@ -126,18 +126,27 @@ int KafkaCgroup::kafka_roundrobin_assignor(kafka_member_t **members,
 			struct list_head *pos;
 			KafkaToppar *toppar;
 
-			list_for_each(pos, &members[next]->toppar_list)
+			int i = 0;
+			for (; i < member_elements; i++)
 			{
-				toppar = list_entry(pos, KafkaToppar, list);
-				if (strcmp(subscriber.get_meta()->get_topic(), toppar->get_topic()) == 0)
-					break;
-				else
-					next++;
+				bool flag = false;
+				list_for_each(pos, &members[next + i]->toppar_list)
+				{
+					toppar = list_entry(pos, KafkaToppar, list);
+					if (strcmp(subscriber.get_meta()->get_topic(), toppar->get_topic()) == 0)
+					{
+						flag = true;
+						break;
+					}
+				}
 
-				if (next >= (int) subscriber.get_member()->size())
-					abort();
+				if (flag)
+					break;
 			}
 
+			if (i >= member_elements)
+				return -1;
+
 			toppar = new KafkaToppar;
 			if (!toppar->set_topic_partition(subscriber.get_meta()->get_topic(),
 											 partition))

src/protocol/KafkaDataTypes.h

@@ -1059,6 +1059,11 @@ public:
 		return &this->ptr->assigned_toppar_list;
 	}
 
+	KafkaToppar *get_assigned_toppar_by_pos(struct list_head *pos)
+	{
+		return list_entry(pos, KafkaToppar, list);
+	}
+
 	void assigned_toppar_rewind();
 
 	KafkaToppar *get_assigned_toppar_next();

src/protocol/KafkaMessage.cc

@@ -1513,7 +1513,7 @@ int KafkaMessage::parse_records(void **buf, size_t *size,
 
 		if (ret > 0)
 		{
-			*size -= msg_size;
+			*size -= msg_set_size;
 			*buf = (char *)*buf + msg_size;
 			return 0;
 		}
@@ -1522,6 +1522,7 @@ int KafkaMessage::parse_records(void **buf, size_t *size,
 	}
 
 	*size -= msg_set_size;
+	*buf = (char *)*buf + msg_size;
 	return 0;
 }
 
@@ -2118,7 +2119,14 @@ int KafkaRequest::encode_produce(struct iovec vectors[], int max)
 			{
 				toppar->record_rollback();
 				toppar->save_record_endpos();
-				record_flag = 1;
+				if (record_flag < 0)
+				{
+					errno = EMSGSIZE;
+					return -1;
+				}
+				else
+					record_flag = 1;
+
 				break;
 			}
 
@@ -3137,9 +3145,6 @@ int KafkaResponse::parse_fetch(void **buf, size_t *size)
 		{
 			CHECK_RET(parse_i32(buf, size, &partition));
 
-			if (partition == 2)
-				partition = 2;
-
 			KafkaToppar *toppar = find_toppar_by_name(topic_name, partition,
 													  &this->toppar_list);
 
@@ -3369,8 +3374,6 @@ static int kafka_cgroup_parse_member(void **buf, size_t *size,
 
 		if (j != topic_cnt)
 			break;
-
-		member++;
 	}
 
 	if (i != member_cnt)