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;