diff --git a/src/client/WFKafkaClient.cc b/src/client/WFKafkaClient.cc
index a8b426da..f30437ae 100644
--- a/src/client/WFKafkaClient.cc
+++ b/src/client/WFKafkaClient.cc
@@ -670,11 +670,12 @@ void ComplexKafkaTask::kafka_parallel_callback(const ParallelWork *pwork)
 {
 	ComplexKafkaTask *t = (ComplexKafkaTask *)pwork->get_context();
 	t->finish = true;
-	t->state = WFT_STATE_SUCCESS;
+	t->state = WFT_STATE_TASK_ERROR;
 	t->error = 0;
 
 	std::pair<int, int> *state_error;
 	bool flag = false;
+	int error = 0;
 	for (size_t i = 0; i < pwork->size(); i++)
 	{
 		state_error = (std::pair<int, int> *)pwork->series_at(i)->get_context();
@@ -687,12 +688,18 @@ void ComplexKafkaTask::kafka_parallel_callback(const ParallelWork *pwork)
 				t->set_meta_status(META_UNINIT);
 				t->lock_status.get_mutex()->unlock();
 			}
-			t->state = state_error->first;
-			t->error = state_error->second;
+			error = state_error->second;
+		}
+		else
+		{
+			t->state = WFT_STATE_SUCCESS;
 		}
 
 		delete state_error;
 	}
+
+	if (t->state != WFT_STATE_SUCCESS)
+		t->error = error;
 }
 
 void ComplexKafkaTask::kafka_process_toppar_offset(KafkaToppar *task_toppar)
@@ -730,7 +737,8 @@ void ComplexKafkaTask::kafka_move_task_callback(__WFKafkaTask *task)
 
 	KafkaTopparList *toppar_list = task->get_resp()->get_toppar_list();
 
-	if (task->get_resp()->get_api_type() == Kafka_Fetch)
+	if (task->get_state() == WFT_STATE_SUCCESS &&
+		task->get_resp()->get_api_type() == Kafka_Fetch)
 	{
 		toppar_list->rewind();
 		KafkaToppar *task_toppar;
diff --git a/src/factory/KafkaTaskImpl.cc b/src/factory/KafkaTaskImpl.cc
index b61d0916..1263c181 100644
--- a/src/factory/KafkaTaskImpl.cc
+++ b/src/factory/KafkaTaskImpl.cc
@@ -515,7 +515,8 @@ bool __ComplexKafkaTask::process_fetch()
 	this->get_resp()->get_toppar_list()->rewind();
 	while ((toppar = this->get_resp()->get_toppar_list()->get_next()) != NULL)
 	{
-		if (toppar->get_error() == KAFKA_OFFSET_OUT_OF_RANGE)
+		if (toppar->get_error() == KAFKA_OFFSET_OUT_OF_RANGE &&
+			toppar->get_offset() >= 0)
 		{
 			toppar->set_offset(KAFKA_OFFSET_OVERFLOW);
 			toppar->set_low_watermark(KAFKA_OFFSET_UNINIT);
@@ -537,8 +538,16 @@ bool __ComplexKafkaTask::process_produce()
 			this->get_req()->set_api_type(Kafka_Produce);
 			return true;
 		}
-		if (toppar->get_error() == KAFKA_NOT_LEADER_FOR_PARTITION)
+
+		switch (toppar->get_error())
 		{
+		case KAFKA_UNKNOWN_TOPIC_OR_PARTITION:
+		case KAFKA_LEADER_NOT_AVAILABLE:
+		case KAFKA_NOT_LEADER_FOR_PARTITION:
+		case KAFKA_BROKER_NOT_AVAILABLE:
+		case KAFKA_REPLICA_NOT_AVAILABLE:
+		case KAFKA_KAFKA_STORAGE_ERROR:
+		case KAFKA_FENCED_LEADER_EPOCH:
 			this->get_req()->set_api_type(Kafka_Metadata);
 			return true;
 		}
@@ -650,13 +659,8 @@ bool __ComplexKafkaTask::finish_once()
 			return false;
 		}
 
-		if (this->get_resp()->get_api_type() == Kafka_Fetch ||
-			this->get_resp()->get_api_type() == Kafka_Produce ||
-			this->get_resp()->get_api_type() == Kafka_ApiVersions)
-		{
-			if (*get_mutable_ctx())
-				(*get_mutable_ctx())(this);
-		}
+		if (*get_mutable_ctx())
+			(*get_mutable_ctx())(this);
 	}
 	else
 	{
diff --git a/src/protocol/KafkaMessage.cc b/src/protocol/KafkaMessage.cc
index 076002ba..27113230 100644
--- a/src/protocol/KafkaMessage.cc
+++ b/src/protocol/KafkaMessage.cc
@@ -3337,7 +3337,10 @@ int KafkaResponse::parse_listoffset(void **buf, size_t *size)
 			{
 				CHECK_RET(parse_i32(buf, size, &offset_cnt));
 				for (int j = 0; j < offset_cnt; ++j)
-					CHECK_RET(parse_i64(buf, size, (int64_t *)&ptr->offset));
+				{
+					CHECK_RET(parse_i64(buf, size, &offset));
+					ptr->offset = offset;
+				}
 
 				ptr->low_watermark = 0;
 			}