From 94fd1368e152be93b0925da92f0a6b055c9bdb1b Mon Sep 17 00:00:00 2001 From: wzl12356 Date: Fri, 24 Dec 2021 18:00:31 +0800 Subject: [PATCH 01/13] kafka: fix bug about cgroup assigning topic --- src/factory/KafkaTaskImpl.cc | 2 + src/protocol/KafkaDataTypes.cc | 17 ++++--- src/protocol/KafkaDataTypes.h | 82 ++++++++++++++++++---------------- src/protocol/KafkaMessage.cc | 37 ++++++++++----- src/protocol/KafkaMessage.h | 15 +++++++ 5 files changed, 99 insertions(+), 54 deletions(-) diff --git a/src/factory/KafkaTaskImpl.cc b/src/factory/KafkaTaskImpl.cc index 4f743a96..62c20d41 100644 --- a/src/factory/KafkaTaskImpl.cc +++ b/src/factory/KafkaTaskImpl.cc @@ -432,6 +432,7 @@ bool __ComplexKafkaTask::process_join_group() if (msg->get_cgroup()->get_error() == KAFKA_MISSING_TOPIC) { this->get_req()->set_api_type(Kafka_Metadata); + this->get_req()->set_alien(); update_metadata_ = true; } else if (msg->get_cgroup()->get_error() == KAFKA_MEMBER_ID_REQUIRED) @@ -478,6 +479,7 @@ bool __ComplexKafkaTask::process_metadata() { KafkaCgroup *cgroup = msg->get_cgroup(); if (cgroup->run_assignor(msg->get_meta_list(), + msg->get_alien_meta_list(), cgroup->get_protocol_name()) < 0) { this->error = errno; diff --git a/src/protocol/KafkaDataTypes.cc b/src/protocol/KafkaDataTypes.cc index 186d288c..0689327c 100644 --- a/src/protocol/KafkaDataTypes.cc +++ b/src/protocol/KafkaDataTypes.cc @@ -208,10 +208,9 @@ bool KafkaMeta::create_partitions(int partition_cnt) return true; } -int KafkaCgroup::run_assignor(KafkaMetaList *meta_list, - const char *protocol_name) +void KafkaCgroup::add_subscriber(KafkaMetaList *meta_list, + std::vector *subscribers) { - std::vector subscribers; meta_list->rewind(); KafkaMeta *meta; @@ -241,13 +240,21 @@ int KafkaCgroup::run_assignor(KafkaMetaList *meta_list, } if (!subscriber.get_member()->empty()) - subscribers.emplace_back(subscriber); + subscribers->emplace_back(subscriber); } +} + +int KafkaCgroup::run_assignor(KafkaMetaList *meta_list, + KafkaMetaList *alien_meta_list, + const char *protocol_name) +{ + std::vector subscribers; + this->add_subscriber(meta_list, &subscribers); + this->add_subscriber(alien_meta_list, &subscribers); struct list_head *pos; kafka_group_protocol_t *protocol; bool flag = false; - list_for_each(pos, this->get_group_protocol()) { protocol = list_entry(pos, kafka_group_protocol_t, list); diff --git a/src/protocol/KafkaDataTypes.h b/src/protocol/KafkaDataTypes.h index 44fc355c..0cc31f10 100644 --- a/src/protocol/KafkaDataTypes.h +++ b/src/protocol/KafkaDataTypes.h @@ -1255,6 +1255,44 @@ private: friend const KafkaMeta *get_meta(const char *topic, KafkaMetaList *meta_list); }; +class KafkaMetaSubscriber +{ +public: + void set_meta(KafkaMeta *meta) + { + this->meta = meta; + } + + const KafkaMeta *get_meta() const + { + return this->meta; + } + + void add_member(kafka_member_t *member) + { + this->member_vec.push_back(member); + } + + const std::vector *get_member() const + { + return &this->member_vec; + } + + static bool cmp(const kafka_member_t *m1, const kafka_member_t *m2) + { + return strcmp(m1->member_id, m2->member_id) < 0; + } + + void sort_by_member() + { + std::sort(this->member_vec.begin(), this->member_vec.end(), cmp); + } + +private: + KafkaMeta *meta; + std::vector member_vec; +}; + class KafkaCgroup { public: @@ -1353,7 +1391,11 @@ public: return this->coordinator; } - int run_assignor(KafkaMetaList *meta_list, const char *protocol_name); + int run_assignor(KafkaMetaList *meta_list, KafkaMetaList *alien_meta_list, + const char *protocol_name); + + void add_subscriber(KafkaMetaList *meta_list, + std::vector *subscribers); static int kafka_range_assignor(kafka_member_t **members, int member_elements, @@ -1466,44 +1508,6 @@ private: friend class KafkaList; }; -class KafkaMetaSubscriber -{ -public: - void set_meta(KafkaMeta *meta) - { - this->meta = meta; - } - - const KafkaMeta *get_meta() const - { - return this->meta; - } - - void add_member(kafka_member_t *member) - { - this->member_vec.push_back(member); - } - - const std::vector *get_member() const - { - return &this->member_vec; - } - - static bool cmp(const kafka_member_t *m1, const kafka_member_t *m2) - { - return strcmp(m1->member_id, m2->member_id) < 0; - } - - void sort_by_member() - { - std::sort(this->member_vec.begin(), this->member_vec.end(), cmp); - } - -private: - KafkaMeta *meta; - std::vector member_vec; -}; - class KafkaBuffer { public: diff --git a/src/protocol/KafkaMessage.cc b/src/protocol/KafkaMessage.cc index 6f2e58b5..862094b1 100644 --- a/src/protocol/KafkaMessage.cc +++ b/src/protocol/KafkaMessage.cc @@ -1614,6 +1614,7 @@ KafkaMessage::KafkaMessage() this->stream = new EncodeStream; this->api_type = Kafka_Unknown; this->cur_size = 0; + this->alien = false; } KafkaMessage::~KafkaMessage() @@ -2431,11 +2432,17 @@ int KafkaRequest::encode_metadata(struct iovec vectors[], int max) else append_i32(this->msgbuf, 0); - this->meta_list.rewind(); + KafkaMetaList *p_meta_list; + if (this->alien) + p_meta_list = &this->alien_meta_list; + else + p_meta_list = &this->meta_list; + + p_meta_list->rewind(); KafkaMeta *meta; int topic_cnt = 0; - while ((meta = this->meta_list.get_next()) != NULL) + while ((meta = p_meta_list->get_next()) != NULL) { append_string(this->msgbuf, meta->get_topic()); ++topic_cnt; @@ -3053,8 +3060,14 @@ int KafkaResponse::parse_metadata(void **buf, size_t *size) if (this->api_version >= 1) CHECK_RET(parse_i32(buf, size, &controller_id)); + KafkaMetaList *p_meta_list; + if (this->alien) + p_meta_list = &this->alien_meta_list; + else + p_meta_list = &this->meta_list; + CHECK_RET(kafka_meta_parse_topic(buf, size, this->api_version, - &this->meta_list, &this->broker_list)); + p_meta_list, &this->broker_list)); return 0; } @@ -3382,6 +3395,7 @@ static bool kafka_meta_find_topic(const std::string& topic_name, static int kafka_cgroup_parse_member(void **buf, size_t *size, KafkaCgroup *cgroup, KafkaMetaList *meta_list, + KafkaMetaList *alien_meta_list, int api_version) { int member_cnt = 0; @@ -3397,7 +3411,7 @@ static int kafka_cgroup_parse_member(void **buf, size_t *size, return -1; kafka_member_t **member = cgroup->get_members(); - std::set missing_topic; + std::set alien_topic; int i; for (i = 0; i < member_cnt; ++i) @@ -3443,7 +3457,7 @@ static int kafka_cgroup_parse_member(void **buf, size_t *size, list_add_tail(toppar->get_list(), &member[i]->toppar_list); if (!kafka_meta_find_topic(topic_name, meta_list)) - missing_topic.insert(topic_name); + alien_topic.insert(topic_name); } if (j != topic_cnt) @@ -3453,7 +3467,7 @@ static int kafka_cgroup_parse_member(void **buf, size_t *size, if (i != member_cnt) return -1; - for (const auto& v : missing_topic) + for (const auto& v : alien_topic) { KafkaMeta *meta = new KafkaMeta; if (!meta->set_topic(v)) @@ -3462,10 +3476,10 @@ static int kafka_cgroup_parse_member(void **buf, size_t *size, return -1; } - meta_list->add_item(std::move(*meta)); + alien_meta_list->add_item(std::move(*meta)); } - if (!missing_topic.empty()) + if (!alien_topic.empty()) cgroup->set_error(KAFKA_MISSING_TOPIC); return 0; @@ -3486,11 +3500,14 @@ int KafkaResponse::parse_joingroup(void **buf, size_t *size) CHECK_RET(parse_string(buf, size, &cgroup->leader_id)); CHECK_RET(parse_string(buf, size, &cgroup->member_id)); CHECK_RET(kafka_cgroup_parse_member(buf, size, &this->cgroup, - &this->meta_list, this->api_version)); + &this->meta_list, + &this->alien_meta_list, + this->api_version)); - if (this->cgroup.is_leader()) + if (cgroup->error != KAFKA_MISSING_TOPIC && this->cgroup.is_leader()) { CHECK_RET(this->cgroup.run_assignor(&this->meta_list, + &this->alien_meta_list, cgroup->protocol_name)); } diff --git a/src/protocol/KafkaMessage.h b/src/protocol/KafkaMessage.h index 1ddb797c..b12aae3d 100644 --- a/src/protocol/KafkaMessage.h +++ b/src/protocol/KafkaMessage.h @@ -93,6 +93,16 @@ public: return &this->meta_list; } + KafkaMetaList *get_alien_meta_list() + { + return &this->alien_meta_list; + } + + void set_alien() + { + this->alien = true; + } + void set_toppar_list(const KafkaTopparList& toppar_list) { this->toppar_list = toppar_list; @@ -127,10 +137,12 @@ public: this->cgroup = msg.cgroup; this->broker = msg.broker; this->meta_list = msg.meta_list; + this->alien_meta_list = msg.alien_meta_list; this->broker_list = msg.broker_list; this->toppar_list = msg.toppar_list; this->sasl = msg.sasl; this->api = msg.api; + this->alien = msg.alien; } void clear_buf() @@ -189,6 +201,7 @@ protected: KafkaCgroup cgroup; KafkaBroker broker; KafkaMetaList meta_list; + KafkaMetaList alien_meta_list; KafkaBrokerList broker_list; KafkaTopparList toppar_list; KafkaBuffer serialized; @@ -205,6 +218,8 @@ protected: kafka_sasl_t *sasl; kafka_api_t *api; + + bool alien; }; class KafkaRequest : public KafkaMessage From 05634ab8b9a7fd5da4b20caf3dcf863b0a0113b1 Mon Sep 17 00:00:00 2001 From: chanchann Date: Sun, 26 Dec 2021 10:10:42 +0800 Subject: [PATCH 02/13] add workflow-config-version.cmake --- CMakeLists.txt | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/CMakeLists.txt b/CMakeLists.txt index 26979c3f..26149465 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -94,6 +94,12 @@ configure_package_config_file( PATH_VARS CONFIG_INC_DIR CONFIG_LIB_DIR ) +write_basic_package_version_file( + ${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}-config-version.cmake + VERSION ${WORKFLOW_VERSION} + COMPATIBILITY AnyNewerVersion +) + install( FILES ${CMAKE_CONFIG_INSTALL_FILE} DESTINATION ${CMAKE_CONFIG_INSTALL_DIR} @@ -101,6 +107,12 @@ install( RENAME ${PROJECT_NAME}-config.cmake ) +install( + FILES ${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}-config-version.cmake + DESTINATION ${CMAKE_CONFIG_INSTALL_DIR} + COMPONENT devel +) + install( FILES ${INCLUDE_HEADERS} ${INCLUDE_KERNEL_HEADERS} DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/${PROJECT_NAME} From 32c21c83feb8b2d7960c9f1ee0407635c2e07849 Mon Sep 17 00:00:00 2001 From: XieHan Date: Wed, 29 Dec 2021 02:20:22 +0800 Subject: [PATCH 03/13] remove HttpMessage::get_output_body_nocopy --- src/protocol/HttpMessage.cc | 24 ------------------------ src/protocol/HttpMessage.h | 2 -- 2 files changed, 26 deletions(-) diff --git a/src/protocol/HttpMessage.cc b/src/protocol/HttpMessage.cc index cf1961b8..6094cde0 100644 --- a/src/protocol/HttpMessage.cc +++ b/src/protocol/HttpMessage.cc @@ -67,30 +67,6 @@ bool HttpMessage::append_output_body_nocopy(const void *buf, size_t size) return false; } -bool HttpMessage::get_output_body_nocopy(struct iovec iov[], int *iovcnt) const -{ - struct HttpMessageBlock *block; - struct list_head *pos; - int i = 0; - - list_for_each(pos, &this->output_body) - { - if (i >= *iovcnt) - { - errno = ENOBUFS; - return false; - } - - block = list_entry(pos, struct HttpMessageBlock, list); - iov[i].iov_base = (void *)block->ptr; - iov[i].iov_len = block->size; - i++; - } - - *iovcnt = i; - return true; -} - void HttpMessage::clear_output_body() { struct HttpMessageBlock *block; diff --git a/src/protocol/HttpMessage.h b/src/protocol/HttpMessage.h index cd147e14..07461b50 100644 --- a/src/protocol/HttpMessage.h +++ b/src/protocol/HttpMessage.h @@ -19,7 +19,6 @@ #ifndef _HTTPMESSAGE_H_ #define _HTTPMESSAGE_H_ -#include #include #include #include @@ -128,7 +127,6 @@ public: * msg->append_output_body_nocopy(body, size); */ bool append_output_body(const void *buf, size_t size); bool append_output_body_nocopy(const void *buf, size_t size); - bool get_output_body_nocopy(struct iovec iov[], int *iovcnt) const; void clear_output_body(); size_t get_output_body_size() const From 78cb4f3a4620a848f4704eab26d9df1e4a717d29 Mon Sep 17 00:00:00 2001 From: Lance Lin Date: Wed, 29 Dec 2021 21:18:46 +0700 Subject: [PATCH 04/13] Updated README.md to include Debian installation instructions --- README.md | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/README.md b/README.md index 3b0b55dc..77aa04e2 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,37 @@ int main() * Uses the `C++11` standard and therefore, it should be compiled with a compiler which supports `C++11`. Does not rely on `boost` or `asio`. * No other dependencies. However, if you need `Kafka` protocol, some compression libraries should be installed, including `lz4`, `zstd` and `snappy`. +### Get started (Debian Linux): +Sogou C++ Workflow has been packaged for Debian. It is currently in Debian sid (unstable) but will eventually be placed into the stable repository. + +In order to access the unstable repository, you will need to edit your /etc/apt/sources.list file. + +sources.list has the format: `deb ` + +Simply add the 'unstable' sub branch to your repo: +~~~~sh +deb http://deb.debian.org/ main contrib non-free + +--> + +deb http://deb.debian.org/ unstable main contrib non-free +~~~~ + +Once that is added, update your repo list and then you should be able to install it: +~~~~sh +sudo apt-get update +~~~~ + +To install the Workflow library for development purposes: +~~~~sh +sudo apt-get install libworkflow-dev +~~~~ + +To install the Workflow library for deployment: +~~~~sh +sudo apt-get install libworkflow1 +~~~~ + ### Get started (Linux, macOS): ~~~sh git clone https://github.com/sogou/workflow From e837260a01f35db4f29c2f45a3d6b0b2ae94b317 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Thu, 30 Dec 2021 12:03:22 +0800 Subject: [PATCH 05/13] Update README.md --- README.md | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 77aa04e2..2d980110 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,14 @@ int main() * Uses the `C++11` standard and therefore, it should be compiled with a compiler which supports `C++11`. Does not rely on `boost` or `asio`. * No other dependencies. However, if you need `Kafka` protocol, some compression libraries should be installed, including `lz4`, `zstd` and `snappy`. +### Get started (Linux, macOS): +~~~sh +git clone https://github.com/sogou/workflow +make +cd tutorial +make +~~~~ + ### Get started (Debian Linux): Sogou C++ Workflow has been packaged for Debian. It is currently in Debian sid (unstable) but will eventually be placed into the stable repository. @@ -84,14 +92,6 @@ To install the Workflow library for deployment: sudo apt-get install libworkflow1 ~~~~ -### Get started (Linux, macOS): -~~~sh -git clone https://github.com/sogou/workflow -make -cd tutorial -make -~~~~ - # Tutorials * Client From e4db13ac3f0d8fd768763069445aba5fd69728ac Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 3 Jan 2022 16:29:39 +0800 Subject: [PATCH 06/13] Update README.md --- README.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/README.md b/README.md index 2d980110..a8896a64 100644 --- a/README.md +++ b/README.md @@ -66,14 +66,10 @@ Sogou C++ Workflow has been packaged for Debian. It is currently in Debian sid ( In order to access the unstable repository, you will need to edit your /etc/apt/sources.list file. -sources.list has the format: `deb ` - Simply add the 'unstable' sub branch to your repo: ~~~~sh deb http://deb.debian.org/ main contrib non-free - --> - deb http://deb.debian.org/ unstable main contrib non-free ~~~~ From 848386749ffc1674888e2bcab3a2900cee20e461 Mon Sep 17 00:00:00 2001 From: wzl12356 Date: Thu, 6 Jan 2022 06:34:59 +0800 Subject: [PATCH 07/13] kafka: set meta uninit when task failed --- src/client/WFKafkaClient.cc | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/src/client/WFKafkaClient.cc b/src/client/WFKafkaClient.cc index f4b5b6bb..303eb12f 100644 --- a/src/client/WFKafkaClient.cc +++ b/src/client/WFKafkaClient.cc @@ -326,6 +326,7 @@ private: int get_node_id(const KafkaToppar *toppar); MetaStatus get_meta_status(); + void set_meta_status(MetaStatus status); std::string get_userinfo() { return this->userinfo; } @@ -673,12 +674,19 @@ void ComplexKafkaTask::kafka_parallel_callback(const ParallelWork *pwork) t->error = 0; std::pair *state_error; - + bool flag = false; for (size_t i = 0; i < pwork->size(); i++) { state_error = (std::pair *)pwork->series_at(i)->get_context(); if (state_error->first != WFT_STATE_SUCCESS) { + if (!flag) + { + flag = true; + t->lock_status.get_mutex()->lock(); + t->set_meta_status(META_UNINIT); + t->lock_status.get_mutex()->unlock(); + } t->state = state_error->first; t->error = state_error->second; } @@ -808,6 +816,14 @@ MetaStatus ComplexKafkaTask::get_meta_status() return ret; } +void ComplexKafkaTask::set_meta_status(MetaStatus status) +{ + this->meta_list.rewind(); + KafkaMeta *meta; + while ((meta = this->meta_list.get_next()) != NULL) + (*this->client->member->meta_map)[meta->get_topic()] = status; +} + void ComplexKafkaTask::dispatch() { __WFKafkaTask *task; From 80ec6110fb615adb38dc307d4be3ed762d789322 Mon Sep 17 00:00:00 2001 From: wzl12356 Date: Thu, 6 Jan 2022 07:27:45 +0800 Subject: [PATCH 08/13] kafka: set meta uninit when task failed --- src/client/WFKafkaClient.cc | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/client/WFKafkaClient.cc b/src/client/WFKafkaClient.cc index 303eb12f..c11430b5 100644 --- a/src/client/WFKafkaClient.cc +++ b/src/client/WFKafkaClient.cc @@ -706,7 +706,7 @@ void ComplexKafkaTask::kafka_process_toppar_offset(KafkaToppar *task_toppar) if (strcmp(toppar->get_topic(), task_toppar->get_topic()) == 0 && toppar->get_partition() == task_toppar->get_partition()) { - if (task_toppar->get_error() == KAFKA_NONE && + if (task_toppar->get_error() == KAFKA_NONE && !task_toppar->reach_high_watermark()) toppar->set_offset(task_toppar->get_offset() + 1); else @@ -818,9 +818,9 @@ MetaStatus ComplexKafkaTask::get_meta_status() void ComplexKafkaTask::set_meta_status(MetaStatus status) { - this->meta_list.rewind(); + this->client_meta_list.rewind(); KafkaMeta *meta; - while ((meta = this->meta_list.get_next()) != NULL) + while ((meta = this->client_meta_list.get_next()) != NULL) (*this->client->member->meta_map)[meta->get_topic()] = status; } From 0be6ff46fdae0c24b79e9ea2d55cd062b9bcc6ce Mon Sep 17 00:00:00 2001 From: wzl12356 Date: Thu, 6 Jan 2022 10:27:45 +0800 Subject: [PATCH 09/13] kafka: set meta uninit when task failed --- src/protocol/KafkaMessage.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/protocol/KafkaMessage.cc b/src/protocol/KafkaMessage.cc index 862094b1..ebc87b7d 100644 --- a/src/protocol/KafkaMessage.cc +++ b/src/protocol/KafkaMessage.cc @@ -44,7 +44,7 @@ namespace protocol #define CHECK_RET(exp) if (exp < 0) return exp #ifndef htonll -static int64_t htonll(int64_t x) +static uint64_t htonll(uint64_t x) { if (1 == htonl(1)) return x; @@ -3500,7 +3500,7 @@ int KafkaResponse::parse_joingroup(void **buf, size_t *size) CHECK_RET(parse_string(buf, size, &cgroup->leader_id)); CHECK_RET(parse_string(buf, size, &cgroup->member_id)); CHECK_RET(kafka_cgroup_parse_member(buf, size, &this->cgroup, - &this->meta_list, + &this->meta_list, &this->alien_meta_list, this->api_version)); From 4a8c1134d7c3f0d3739aae60ac69381921596e77 Mon Sep 17 00:00:00 2001 From: wzl12356 Date: Thu, 6 Jan 2022 17:56:22 +0800 Subject: [PATCH 10/13] kafka: add enum to MetaStatus --- src/client/WFKafkaClient.cc | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/client/WFKafkaClient.cc b/src/client/WFKafkaClient.cc index c11430b5..a8b426da 100644 --- a/src/client/WFKafkaClient.cc +++ b/src/client/WFKafkaClient.cc @@ -125,7 +125,7 @@ public: this->broker_list = new KafkaBrokerList; this->lock_status = new KafkaLockStatus; this->broker_map = new KafkaBrokerMap; - this->meta_map = new std::map; + this->meta_map = new std::map; } ~KafkaMember() @@ -149,7 +149,7 @@ public: KafkaBrokerList *broker_list; KafkaBrokerMap *broker_map; KafkaLockStatus *lock_status; - std::map *meta_map; + std::map *meta_map; private: std::atomic *ref; @@ -325,8 +325,8 @@ private: int get_node_id(const KafkaToppar *toppar); - MetaStatus get_meta_status(); - void set_meta_status(MetaStatus status); + enum MetaStatus get_meta_status(); + void set_meta_status(enum MetaStatus status); std::string get_userinfo() { return this->userinfo; } @@ -789,11 +789,11 @@ void ComplexKafkaTask::parse_query() } } -MetaStatus ComplexKafkaTask::get_meta_status() +enum MetaStatus ComplexKafkaTask::get_meta_status() { this->meta_list.rewind(); KafkaMeta *meta; - MetaStatus ret = META_INITED; + enum MetaStatus ret = META_INITED; while ((meta = this->meta_list.get_next()) != NULL) { switch((*this->client->member->meta_map)[meta->get_topic()]) @@ -816,7 +816,7 @@ MetaStatus ComplexKafkaTask::get_meta_status() return ret; } -void ComplexKafkaTask::set_meta_status(MetaStatus status) +void ComplexKafkaTask::set_meta_status(enum MetaStatus status) { this->client_meta_list.rewind(); KafkaMeta *meta; From 4f8f789b83be169353967b86c0b161e36470fc3a Mon Sep 17 00:00:00 2001 From: XieHan Date: Wed, 19 Jan 2022 20:39:06 +0800 Subject: [PATCH 11/13] fix CMakeLists.txt of tutorial --- tutorial/CMakeLists.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tutorial/CMakeLists.txt b/tutorial/CMakeLists.txt index b648166a..830319da 100644 --- a/tutorial/CMakeLists.txt +++ b/tutorial/CMakeLists.txt @@ -76,7 +76,7 @@ endif() if (KAFKA STREQUAL "y") add_executable("kafka_cli" "tutorial-13-kafka_cli.cc") - target_link_libraries("kafka_cli" wfkafka workflow z snappy lz4 zstd rt) + target_link_libraries("kafka_cli" wfkafka ${WORKFLOW_LIB} z snappy lz4 zstd) endif () set(DIR10 tutorial-10-user_defined_protocol) From b624d8992a46947038593bc51d78c39b238f8cd4 Mon Sep 17 00:00:00 2001 From: XieHan Date: Wed, 19 Jan 2022 21:03:29 +0800 Subject: [PATCH 12/13] fix shared library problem on Mac --- src/CMakeLists.txt | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 654851b5..75170538 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -55,6 +55,10 @@ add_library( $ ) +if (APPLE) + target_link_libraries(${SHARED_LIB_NAME} ssl crypto pthread) +endif () + set_target_properties(${STATIC_LIB_NAME} PROPERTIES OUTPUT_NAME ${PROJECT_NAME}) set_target_properties(${SHARED_LIB_NAME} PROPERTIES OUTPUT_NAME ${PROJECT_NAME}) From 1cfa69f27c69929b66bb15463e9bd36538a996a6 Mon Sep 17 00:00:00 2001 From: XieHan Date: Wed, 19 Jan 2022 21:57:49 +0800 Subject: [PATCH 13/13] Donot generate wfkafka shared lib on macOS --- src/CMakeLists.txt | 32 +++++++++++++++++++++----------- 1 file changed, 21 insertions(+), 11 deletions(-) diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 75170538..214c430b 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -56,7 +56,7 @@ add_library( ) if (APPLE) - target_link_libraries(${SHARED_LIB_NAME} ssl crypto pthread) + target_link_libraries(${SHARED_LIB_NAME} OpenSSL::SSL OpenSSL::Crypto pthread) endif () set_target_properties(${STATIC_LIB_NAME} PROPERTIES OUTPUT_NAME ${PROJECT_NAME}) @@ -68,7 +68,6 @@ if (KAFKA STREQUAL "y") add_dependencies(protocol_kafka LINK_HEADERS) add_dependencies(factory_kafka LINK_HEADERS) set(KAFKA_STATIC_LIB_NAME "wfkafka-static") - set(KAFKA_SHARED_LIB_NAME "wfkafka-shared") add_library( ${KAFKA_STATIC_LIB_NAME} STATIC $ @@ -76,15 +75,18 @@ if (KAFKA STREQUAL "y") $ $ ) - add_library( - ${KAFKA_SHARED_LIB_NAME} SHARED - $ - $ - $ - $ - ) set_target_properties(${KAFKA_STATIC_LIB_NAME} PROPERTIES OUTPUT_NAME "wfkafka") - set_target_properties(${KAFKA_SHARED_LIB_NAME} PROPERTIES OUTPUT_NAME "wfkafka") + if (NOT APPLE) + set(KAFKA_SHARED_LIB_NAME "wfkafka-shared") + add_library( + ${KAFKA_SHARED_LIB_NAME} SHARED + $ + $ + $ + $ + ) + set_target_properties(${KAFKA_SHARED_LIB_NAME} PROPERTIES OUTPUT_NAME "wfkafka") + endif () endif () install( @@ -96,10 +98,18 @@ install( if (KAFKA STREQUAL "y") install( - TARGETS ${KAFKA_STATIC_LIB_NAME} ${KAFKA_SHARED_LIB_NAME} + TARGETS ${KAFKA_STATIC_LIB_NAME} ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR} COMPONENT devel ) + if (NOT APPLE) + install( + TARGETS ${KAFKA_SHARED_LIB_NAME} + ARCHIVE + DESTINATION ${CMAKE_INSTALL_LIBDIR} + COMPONENT devel + ) + endif () endif ()