DDLWorker thread checks in a loop for new tasks by calling ZooKeeper GetChildren with a watch callback.

https://github.com/ClickHouse/ClickHouse/blob/5b0bc8a7fbf846cb45012e8009d58341656bf12c/src/Interpreters/DDLWorker.cpp#L374

Due to a recent change from #25373 instead of blocking on new events we retry the loop every 10 seconds instead of when watch callback is triggered.

https://github.com/ClickHouse/ClickHouse/blob/5b0bc8a7fbf846cb45012e8009d58341656bf12c/src/Interpreters/DDLWorker.cpp#L1140-L1142

This causes a zookeeper watch callback leak. It is probably not very important as all that happens is that a new std::function is added to the watches list for the ddl worker path every 10 seconds. What is 32 bytes (sizeof(std::function)?) every 10 seconds?

However, it is annoying to see that on metrics. The metric is somewhat useful for catching bugs and this isn't an important one.

I understand that we could probably fix this "temporary fix" in DDLWorker and remove the tryWait + retrying, but that pattern was seen in other places and is useful for implementing a non-blocking watch, also used in `waitForTableReplicaToProcessLogEntry`.

Is there a laconic fix for that? Maybe we could somehow remove the callback when not needed/before retry? Maybe we could change the behaviour slightly and add watchcallback only if it wasn't already added (this wouldn't work for getChildren which wraps a Poco Event in a watch)? Anything else beside just ignoring this?

---

I understand that all the callbacks will be cleared when the event is triggered.
}
    else
    {
        
        String last_task_name;
        if (!current_tasks.empty())
            last_task_name = current_tasks.back()->entry_name;
        if (last_skipped_entry_name && last_task_name < *last_skipped_entry_name)
            last_task_name = *last_skipped_entry_name;
        begin_node = std::upper_bound(queue_nodes.begin(), queue_nodes.end(), last_task_name);
    }
    if (begin_node == queue_nodes.end())
        LOG_DEBUG(log, "No tasks to schedule");
    else
        LOG_DEBUG(log, "Will schedule {} tasks starting from {}", std::distance(begin_node, queue_nodes.end()), *begin_node);
    
    
    
    [[maybe_unused]] bool have_no_tasks_info = !first_failed_task_name && current_tasks.empty() && !last_skipped_entry_name;
    assert(have_no_tasks_info || queue_nodes.end() == std::find_if(queue_nodes.begin(), queue_nodes.end(), [&](const String & entry_name)
    {
        
        String reason;
        auto task = initAndCheckTask(entry_name, reason, zookeeper);
        bool maybe_currently_processing = current_tasks.end() != std::find_if(current_tasks.begin(), current_tasks.end(), [&](const auto & t)
        {
            return t->entry_name == entry_name;
        });
        
        if (begin_node == queue_nodes.end() || entry_name < *begin_node)