| 435 | } |
| 436 | |
| 437 | comm::RetCode SchedulerMgr::ReloadConsumerConfig(const int topic_id, TopicData &topic_data, |
| 438 | TopicLock &topic_lock, const uint64_t now, |
| 439 | bool &modified) { |
| 440 | modified = false; |
| 441 | |
| 442 | uint64_t conf_last_mod_time{config::GlobalConfig::GetThreadInstance()->GetLastModTime(topic_id)}; |
| 443 | if (conf_last_mod_time == topic_data.conf_last_mod_time) { |
| 444 | QLVerb("topic %d not outdated skip new_mod_time %lu old_mod_time %lu", |
| 445 | topic_id, conf_last_mod_time, topic_data.conf_last_mod_time); |
| 446 | |
| 447 | return comm::RetCode::RET_OK; |
| 448 | } else { |
| 449 | QLVerb("topic %d outdated new_mod_time %lu old_mod_time %lu", |
| 450 | topic_id, conf_last_mod_time, topic_data.conf_last_mod_time); |
| 451 | } |
| 452 | |
| 453 | shared_ptr<const config::ConsumerConfig> consumer_config; |
| 454 | comm::RetCode ret{config::GlobalConfig::GetThreadInstance()-> |
| 455 | GetConsumerConfig(topic_id, consumer_config)}; |
| 456 | if (comm::RetCode::RET_OK != ret) { |
| 457 | QLErr("GetConsumerConfig err %d", ret); |
| 458 | |
| 459 | return ret; |
| 460 | } |
| 461 | |
| 462 | map<uint64_t, int> consumer_addr2scale_map; |
| 463 | ret = GetConsumerConfigDetail(topic_id, consumer_addr2scale_map, consumer_config); |
| 464 | if (comm::RetCode::RET_OK != ret) { |
| 465 | QLErr("topic %d GetConsumerConfigDetail err %d", topic_id, ret); |
| 466 | } |
| 467 | |
| 468 | auto &consumer_addr2info_map(topic_data.consumer_addr2info_map); |
| 469 | |
| 470 | // check map update |
| 471 | bool modify{false}; |
| 472 | { |
| 473 | comm::utils::RWLock rwlock_read(topic_lock.rwlock, comm::utils::RWLock::LockMode::READ); |
| 474 | if (consumer_addr2info_map.size() != consumer_addr2scale_map.size()) { |
| 475 | modify = true; |
| 476 | } else { |
| 477 | for (auto &kv : consumer_addr2scale_map) { |
| 478 | auto it(consumer_addr2info_map.find(kv.first)); |
| 479 | if (consumer_addr2info_map.end() == it) { |
| 480 | modify = true; |
| 481 | |
| 482 | break; |
| 483 | } |
| 484 | } |
| 485 | } |
| 486 | } |
| 487 | |
| 488 | // map updated |
| 489 | if (modify) { |
| 490 | comm::utils::RWLock rwlock_write(topic_lock.rwlock, comm::utils::RWLock::LockMode::WRITE); |
| 491 | |
| 492 | // add |
| 493 | for (auto &&kv : consumer_addr2scale_map) { |
| 494 | if (consumer_addr2info_map.end() == consumer_addr2info_map.find(kv.first)) { |
nothing calls this directly
no test coverage detected