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

kafka: change return type for init and deinit

Browse Source
This commit is contained in:
zhuleiwang
2022-04-22 11:30:10 +08:00
parent 7bbe8daac3
commit 089e64f656
2 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/WFKafkaClient.cc
+6 -3
View File
@@ -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<std::string> 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,
src/client/WFKafkaClient.h
+3 -3
View File
@@ -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