Method initializeMainThread

src/Databases/DatabaseReplicatedWorker.cpp:78–142

}

bool DatabaseReplicatedDDLWorker::initializeMainThread()
{
    {
        std::lock_guard lock(initialization_duration_timer_mutex);
        initialization_duration_timer.emplace();
        initialization_duration_timer->start();
    }

    while (!stop_flag)
    {
        try
        {
            chassert(!database->is_probably_dropped);
            auto zookeeper = getAndSetZooKeeper();
            if (database->is_readonly)
                database->tryConnectToZooKeeperAndInitDatabase(LoadingStrictnessLevel::ATTACH);
            if (database->is_probably_dropped)
            {
                /// The flag was set in tryConnectToZooKeeperAndInitDatabase
                LOG_WARNING(log, "Exiting main thread, because the database was probably dropped");
                /// NOTE It will not stop cleanup thread until DDLWorker::shutdown() call (cleanup thread will just do nothing)
                break;
            }

            if (database->db_settings[DatabaseReplicatedSetting::max_retries_before_automatic_recovery]
                && database->db_settings[DatabaseReplicatedSetting::max_retries_before_automatic_recovery] <= subsequent_errors_count)
            {
                String current_task_name;
                {
                    std::unique_lock lock{mutex};
                    current_task_name = current_task;
                }
                LOG_WARNING(log, "Database got stuck at processing task {}: it failed {} times in a row with the same error. "
                                 "Will reset digest to mark our replica as lost, and trigger recovery from the most up-to-date metadata "
                                 "from ZooKeeper. See max_retries_before_automatic_recovery setting. The error: {}",
                            current_task, subsequent_errors_count.load(), last_unexpected_error);

                String digest_str;
                zookeeper->tryGet(database->replica_path + "/digest", digest_str);
                LOG_WARNING(log, "Resetting digest from {} to {}", digest_str, FORCE_AUTO_RECOVERY_DIGEST);
                zookeeper->trySet(database->replica_path + "/digest", FORCE_AUTO_RECOVERY_DIGEST);
            }

            initializeReplication();
            initialized = true;
            {
                std::lock_guard lock(initialization_duration_timer_mutex);
                initialization_duration_timer.reset();
            }
            return true;
        }
        catch (...)
        {
            tryLogCurrentException(log, fmt::format("Error on initialization of {}", database->getDatabaseName()));
            queue_updated_event->tryWait(5000);
        }
    }