diff --git a/src/client/WFKafkaClient.cc b/src/client/WFKafkaClient.cc index b55b93ff..4f010db1 100644 --- a/src/client/WFKafkaClient.cc +++ b/src/client/WFKafkaClient.cc @@ -1480,7 +1480,7 @@ SubTask *WFKafkaTask::done() return series->pop(); } -void WFKafkaClient::init(const std::string& broker) +int WFKafkaClient::init(const std::string& broker) { std::vector broker_hosts; std::string::size_type pos = broker.find(','); @@ -1514,21 +1514,24 @@ void WFKafkaClient::init(const std::string& broker) this->member = new KafkaMember; this->member->broker_hosts = std::move(broker_hosts); + return 0; } -void WFKafkaClient::init(const std::string& broker, const std::string& group) +int WFKafkaClient::init(const std::string& broker, const std::string& group) { this->init(broker); this->member->cgroup.set_group(group); this->member->status |= KAFKA_CGROUP_INIT | KAFKA_HEARTBEAT_INIT; + return 0; } -void WFKafkaClient::deinit() +int WFKafkaClient::deinit() { this->member->mutex.lock(); this->member->status |= KAFKA_DEINIT; this->member->decref(); this->member->mutex.unlock(); + return 0; } WFKafkaTask *WFKafkaClient::create_kafka_task(const std::string& query, diff --git a/src/client/WFKafkaClient.h b/src/client/WFKafkaClient.h index bf006d05..f36dd458 100644 --- a/src/client/WFKafkaClient.h +++ b/src/client/WFKafkaClient.h @@ -131,11 +131,11 @@ public: // example: kafka://kafka.sogou // example: kafka.sogou:9090 // example: kafka://10.160.23.23:9000,10.123.23.23,kafka://kafka.sogou - void init(const std::string& broker_url); + int init(const std::string& broker_url); - void init(const std::string& broker_url, const std::string& group); + int init(const std::string& broker_url, const std::string& group); - void deinit(); + int deinit(); // example: topic=xxx&topic=yyy&api=fetch // example: api=commit