GithubStar/workflow
1
0
Fork 0
mirror of https://github.com/sogou/workflow.git synced 2026-10-27 02:23:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #809 from wzl12356/master

Browse Source 
kafka: fix offset not continuous when fetch
This commit is contained in:
xiehan
2022-03-18 15:59:06 +08:00
committed by GitHub
parent 1bf6f89744 3b3d9ae073
commit b2fa6df47c
3 changed files with 29 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/WFKafkaClient.cc
+12 -4
View File
@@ -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;
src/factory/KafkaTaskImpl.cc
+13 -9
View File
@@ -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
{
src/protocol/KafkaMessage.cc
+4 -1
View File
@@ -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;
}