From 12f8a4881878e6d0ba94576e3f31b2c30e358ae8 Mon Sep 17 00:00:00 2001 From: wzl12356 Date: Wed, 22 Dec 2021 15:00:58 +0800 Subject: [PATCH] kafka: fix cgroup bug --- src/factory/KafkaTaskImpl.cc | 7 ++++--- src/protocol/KafkaMessage.cc | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/src/factory/KafkaTaskImpl.cc b/src/factory/KafkaTaskImpl.cc index ee8b3dd5..4f743a96 100644 --- a/src/factory/KafkaTaskImpl.cc +++ b/src/factory/KafkaTaskImpl.cc @@ -421,9 +421,10 @@ bool __ComplexKafkaTask::process_join_group() KafkaResponse *msg = this->get_resp(); if (!msg->get_cgroup()->get_coordinator()->is_to_addr()) { - const struct sockaddr *paddr; - socklen_t addrlen; - msg->get_broker()->get_broker_addr(&paddr, &addrlen); + struct sockaddr_storage addr; + socklen_t addrlen = sizeof addr; + const struct sockaddr *paddr = (const struct sockaddr *)&addr; + this->get_peer_addr((struct sockaddr *)&addr, &addrlen); msg->get_cgroup()->get_coordinator()->set_broker_addr(paddr, addrlen); msg->get_cgroup()->get_coordinator()->set_to_addr(1); } diff --git a/src/protocol/KafkaMessage.cc b/src/protocol/KafkaMessage.cc index 2c45c078..6f2e58b5 100644 --- a/src/protocol/KafkaMessage.cc +++ b/src/protocol/KafkaMessage.cc @@ -1135,7 +1135,7 @@ static int parse_varint_u64(void **buf, size_t *size, uint64_t *val) do { - if (*size - 1 == 0) + if (*size == 0) { *size = org_size; errno = EBADMSG;