diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 56749470..34b94689 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,7 +30,7 @@ jobs: name: fedora runs-on: ubuntu-latest container: - image: fedora:35 + image: fedora:latest steps: - uses: actions/checkout@v3 diff --git a/docs/bugs.md b/docs/bugs.md index 23fb14f1..f057de7b 100644 --- a/docs/bugs.md +++ b/docs/bugs.md @@ -20,14 +20,15 @@ void f() auto *task = WFTaskFactory::create_http_task("https://sogou/index.html", 0, 0, nullptr); } ~~~ -这时候http request里的Host必然填写的是"sogou"。此时如果sogou是一个upstream名,指向域名www.sogou.com。并且我们开启了TLS SNI,那么SNI server name信息就是wwww.sogou.com,与http header里的Host是不一致的,会导致SSL错误。 +这时候http request里的Host必然填写的是"sogou"。此时如果sogou是一个upstream名,指向域名www.sogou.com。并且我们开启了TLS SNI,那么SNI server name信息就是www.sogou.com,与http header里的Host是不一致的,会导致SSL错误。 要解决这个问题,用户可以在通过设置prepare函数,在发送请求前修改Host,让它与最终URL里的一致: ~~~cpp void f(); { auto *task = WFTaskFactory::create_http_task("https://sogou/index.html", 0, 0, nullptr); static_cast *>(task)->set_prepare([](WFHttpTask *task){ - task->get_req()->set_header_pair("Host", task->get_current_uri()->host); // 这里得到实际uri里的host。 + auto *t = static_cast *>(task); + task->get_req()->set_header_pair("Host", t->get_current_uri()->host); // 这里得到实际uri里的host。 }); } ~~~ diff --git a/src/client/WFKafkaClient.cc b/src/client/WFKafkaClient.cc index 9c897db2..147f8b68 100644 --- a/src/client/WFKafkaClient.cc +++ b/src/client/WFKafkaClient.cc @@ -1199,6 +1199,9 @@ bool KafkaClientTask::add_toppar(const KafkaToppar& toppar) } new_toppar.set_offset(toppar.get_offset()); + new_toppar.set_offset_timestamp(toppar.get_offset_timestamp()); + new_toppar.set_low_watermark(toppar.get_low_watermark()); + new_toppar.set_high_watermark(toppar.get_high_watermark()); this->toppar_list.add_item(new_toppar); this->meta_list.add_item(tmp); @@ -1229,6 +1232,9 @@ bool KafkaClientTask::add_toppar(const KafkaToppar& toppar) } new_toppar.set_offset(toppar.get_offset()); + new_toppar.set_offset_timestamp(toppar.get_offset_timestamp()); + new_toppar.set_low_watermark(toppar.get_low_watermark()); + new_toppar.set_high_watermark(toppar.get_high_watermark()); this->toppar_list.add_item(new_toppar); this->meta_list.add_item(*meta); @@ -1440,7 +1446,9 @@ int KafkaClientTask::arrange_fetch() return -1; new_toppar.set_offset(toppar->get_offset()); + new_toppar.set_offset_timestamp(toppar->get_offset_timestamp()); new_toppar.set_low_watermark(toppar->get_low_watermark()); + new_toppar.set_high_watermark(toppar->get_high_watermark()); this->toppar_list_map[node_id].add_item(std::move(new_toppar)); } } diff --git a/src/factory/KafkaTaskImpl.cc b/src/factory/KafkaTaskImpl.cc index 57d79eaa..43bf0792 100644 --- a/src/factory/KafkaTaskImpl.cc +++ b/src/factory/KafkaTaskImpl.cc @@ -206,62 +206,53 @@ CommMessageOut *__ComplexKafkaTask::message_out() KafkaConnectionInfo *conn_info = (KafkaConnectionInfo *)this->get_connection()->get_context(); - this->get_req()->set_api(&conn_info->api); + KafkaRequest *req = this->get_req(); + req->set_api(&conn_info->api); - if (this->get_req()->get_api_type() == Kafka_Fetch || - this->get_req()->get_api_type() == Kafka_ListOffsets) + if (req->get_api_type() == Kafka_Fetch || + req->get_api_type() == Kafka_ListOffsets) { - KafkaRequest *req = this->get_req(); - req->get_toppar_list()->rewind(); + KafkaTopparList *req_toppar_lst = req->get_toppar_list(); KafkaToppar *toppar; KafkaTopparList toppar_list; bool flag = false; + long long cfg_ts = req->get_config()->get_offset_timestamp(); + long long tp_ts; - while ((toppar = req->get_toppar_list()->get_next()) != NULL) + req_toppar_lst->rewind(); + while ((toppar = req_toppar_lst->get_next()) != NULL) { - if (toppar->get_low_watermark() < 0) - toppar->set_offset_timestamp(KAFKA_TIMESTAMP_EARLIEST); - else if (toppar->get_high_watermark() < 0) - toppar->set_offset_timestamp(KAFKA_TIMESTAMP_LATEST); - else if (toppar->get_offset() == KAFKA_OFFSET_UNINIT) + tp_ts = toppar->get_offset_timestamp(); + if (tp_ts == KAFKA_TIMESTAMP_UNINIT) + tp_ts = cfg_ts; + + if (toppar->get_offset() == KAFKA_OFFSET_UNINIT) { - long long conf_ts = - this->get_req()->get_config()->get_offset_timestamp(); - if (conf_ts == KAFKA_TIMESTAMP_EARLIEST) - { + if (tp_ts == KAFKA_TIMESTAMP_EARLIEST) toppar->set_offset(toppar->get_low_watermark()); - continue; - } - else if (conf_ts == KAFKA_TIMESTAMP_LATEST) + else if (tp_ts < 0) { toppar->set_offset(toppar->get_high_watermark()); - continue; - } - else - { - toppar->set_offset_timestamp(conf_ts); + tp_ts = KAFKA_TIMESTAMP_LATEST; } } else if (toppar->get_offset() == KAFKA_OFFSET_OVERFLOW) { - if (this->get_req()->get_config()->get_offset_timestamp() == - KAFKA_TIMESTAMP_EARLIEST) - { + if (tp_ts == KAFKA_TIMESTAMP_EARLIEST) toppar->set_offset(toppar->get_low_watermark()); - } else { toppar->set_offset(toppar->get_high_watermark()); + tp_ts = KAFKA_TIMESTAMP_LATEST; } - continue; - } - else - { - continue; } - toppar_list.add_item(*toppar); - flag = true; + if (toppar->get_offset() < 0) + { + toppar->set_offset_timestamp(tp_ts); + toppar_list.add_item(*toppar); + flag = true; + } } if (flag) @@ -523,8 +514,7 @@ bool __ComplexKafkaTask::process_fetch() this->get_resp()->get_toppar_list()->rewind(); while ((toppar = this->get_resp()->get_toppar_list()->get_next()) != NULL) { - if (toppar->get_error() == KAFKA_OFFSET_OUT_OF_RANGE && - toppar->get_high_watermark() - toppar->get_low_watermark() > 0) + if (toppar->get_error() == KAFKA_OFFSET_OUT_OF_RANGE) { toppar->set_offset(KAFKA_OFFSET_OVERFLOW); toppar->set_low_watermark(KAFKA_OFFSET_UNINIT);