Method waitForProcessingQueue

src/Storages/StorageReplicatedMergeTree.cpp:9861–9912

}

bool StorageReplicatedMergeTree::waitForProcessingQueue(UInt64 max_wait_milliseconds, SyncReplicaMode sync_mode, std::unordered_set<String> source_replicas)
{
    auto component_guard = Coordination::setCurrentComponent("StorageReplicatedMergeTree::waitForProcessingQueue");
    /// Let's fetch new log entries firstly
    queue.pullLogsToQueue(getZooKeeperAndAssertNotReadonly(), {}, ReplicatedMergeTreeQueue::SYNC);

    if (sync_mode == SyncReplicaMode::PULL)
        return true;

    /// This is significant, because the execution of this task could be delayed at BackgroundPool.
    /// And we force it to be executed.
    background_operations_assignee.trigger();

    std::unordered_set<String> wait_for_ids;
    std::atomic_bool was_interrupted = false;

    Poco::Event target_entry_event;
    auto callback = [this, &target_entry_event, &wait_for_ids, &was_interrupted, sync_mode]
        (size_t new_queue_size, const String * removed_log_entry_id)
    {
        if (partial_shutdown_called)
        {
            was_interrupted = true;
            target_entry_event.set();
            return;
        }

        if (sync_mode == SyncReplicaMode::STRICT)
        {
            /// Wait for queue to become empty
            if (new_queue_size == 0)
                target_entry_event.set();
            return;
        }

        if (removed_log_entry_id)
            wait_for_ids.erase(*removed_log_entry_id);

        if (wait_for_ids.empty())
            target_entry_event.set();
    };

    const auto handler = queue.addSubscriber(std::move(callback), wait_for_ids, sync_mode, source_replicas);

    if (!target_entry_event.tryWait(max_wait_milliseconds))
        return false;

    if (was_interrupted)
        throw Exception(ErrorCodes::ABORTED, "Shutdown is called for table");

    return true;
}

bool StorageReplicatedMergeTree::dropPartImpl(
    zkutil::ZooKeeperPtr & zookeeper, String part_name, LogEntry & entry, bool detach, bool throw_if_noop)