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.
task_it = current_tasks.erase(task_it);
            }
        }
    }
    Strings queue_nodes = zookeeper->getChildren(queue_dir, &queue_node_stat, queue_updated_event);
    size_t size_before_filtering = queue_nodes.size();
    filterAndSortQueueNodes(queue_nodes);
    
    LOG_TRACE(log, "scheduleTasks: initialized={}, size_before_filtering={}, queue_size={}, "
                   "entries={}..{}, "
                   "first_failed_task_name={}, current_tasks_size={}, "
                   "last_current_task={}, "
                   "last_skipped_entry_name={}",
                   initialized, size_before_filtering, queue_nodes.size(),
                   queue_nodes.empty() ? "none" : queue_nodes.front(), queue_nodes.empty() ? "none" : queue_nodes.back(),
                   first_failed_task_name ? *first_failed_task_name : "none", current_tasks.size(),
                   current_tasks.empty() ? "none" : current_tasks.back()->entry_name,
                   last_skipped_entry_name ? *last_skipped_entry_name : "none");
    if (max_tasks_in_queue < queue_nodes.size())
        cleanup_event->set();
    
    
    
    
    auto begin_node = queue_nodes.begin();
    if (first_failed_task_name)
    {
        
        assert(reinitialized);
        begin_node = std::lower_bound(queue_nodes.begin(), queue_nodes.end(), first_failed_task_name);