| 424 | |
| 425 | |
| 426 | void ReplicatedMergeTreeQueue::removeProcessedEntry(zkutil::ZooKeeperPtr zookeeper, LogEntryPtr & entry) |
| 427 | { |
| 428 | std::optional<time_t> min_unprocessed_insert_time_changed; |
| 429 | std::optional<time_t> max_processed_insert_time_changed; |
| 430 | |
| 431 | bool found = false; |
| 432 | bool need_remove_from_zk = true; |
| 433 | size_t queue_size = 0; |
| 434 | |
| 435 | /// First remove from memory then from ZooKeeper |
| 436 | { |
| 437 | std::unique_lock lock(state_mutex); |
| 438 | if (entry->removed_by_other_entry) |
| 439 | { |
| 440 | need_remove_from_zk = false; |
| 441 | queue_size = queue.size(); |
| 442 | } |
| 443 | else |
| 444 | { |
| 445 | /// Remove the job from the queue in the RAM. |
| 446 | /// You can not just refer to a pre-saved iterator, because someone else might be able to delete the task. |
| 447 | /// Why do we view the queue from the end? |
| 448 | /// - because the task for execution first is moved to the end of the queue, so that in case of failure it remains at the end. |
| 449 | for (Queue::iterator it = queue.end(); it != queue.begin();) |
| 450 | { |
| 451 | --it; |
| 452 | |
| 453 | if (*it == entry) |
| 454 | { |
| 455 | found = true; |
| 456 | updateStateOnQueueEntryRemoval( |
| 457 | entry, /* is_successful = */ true, |
| 458 | min_unprocessed_insert_time_changed, max_processed_insert_time_changed, lock); |
| 459 | |
| 460 | queue.erase(it); |
| 461 | queue_size = queue.size(); |
| 462 | break; |
| 463 | } |
| 464 | } |
| 465 | } |
| 466 | } |
| 467 | |
| 468 | if (!found && need_remove_from_zk) |
| 469 | throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find {} in the memory queue. It is a bug. Entry: {}", |
| 470 | entry->znode_name, entry->toString()); |
| 471 | |
| 472 | notifySubscribers(queue_size); |
| 473 | |
| 474 | if (!need_remove_from_zk) |
| 475 | return; |
| 476 | |
| 477 | auto code = zookeeper->tryRemove(fs::path(replica_path) / "queue" / entry->znode_name); |
| 478 | if (code != Coordination::Error::ZOK) |
| 479 | LOG_ERROR(log, "Couldn't remove {}/queue/{}: {}. This shouldn't happen often.", replica_path, entry->znode_name, Coordination::errorMessage(code)); |
| 480 | |
| 481 | updateTimesInZooKeeper(zookeeper, min_unprocessed_insert_time_changed, max_processed_insert_time_changed); |
| 482 | } |
| 483 | |