Repository: ClickHouse/ClickHouse
File: src/Storages/RabbitMQ/StorageRabbitMQ.cpp (C++)

Code before:

    for (auto & consumer : consumers_ref)
        consumer.lock()->stop();

    LOG_TRACE(log, "Deactivating background tasks");

    /// In case it has not yet been able to setup connection;
    deactivateTask(init_task, true, false);

    /// The order of deactivating tasks is important: wait for streamingToViews() func to finish and
    /// then wait for background event loop to finish.
    deactivateTask(streaming_task, true, false);
    deactivateTask(looping_task, true, true);

    LOG_TRACE(log, "Cleaning up RabbitMQ after table usage");

    /// Just a paranoid try catch, it is not actually needed.
    try
    {
        for (auto & consumer : consumers_ref)
            consumer.lock()->closeConnections();

        if (drop_table)
        {
            cleanupRabbitMQ();
            StreamingStorageRegistry::instance().unregisterTable(getStorageID(), /* if_exists */ true);
        }

        /// It is important to close connection here - before removing consumers, because
        /// it will finish and clean callbacks, which might use those consumers data.
        connection->disconnect();

        for (size_t i = 0; i < num_created_consumers; ++i)
            popConsumer();
    }
    catch (...)
    {
        tryLogCurrentException(log);
    }

    LOG_TRACE(log, "Shutdown finished");
}

void StorageRabbitMQ::renameInMemory(const StorageID & new_table_id)
{
    const auto prev_storage_id = getStorageID();
    IStorage::renameInMemory(new_table_id);
    StreamingStorageRegistry::instance().renameTable(prev_storage_id, getStorageID());
}



Review comment (suggestion):
We could get an exception in `cleanupRabbitMQ` and then `unregister` will never be called, let's make sure it is called regardless of exception above?

Also why do we unregister only for `drop_table`? I guess this will cause an issue with `detach table` and then attach it back

Code after:

    for (auto & consumer : consumers_ref)
        consumer.lock()->stop();

    LOG_TRACE(log, "Deactivating background tasks");

    /// In case it has not yet been able to setup connection;
    deactivateTask(init_task, true, false);

    /// The order of deactivating tasks is important: wait for streamingToViews() func to finish and
    /// then wait for background event loop to finish.
    deactivateTask(streaming_task, true, false);
    deactivateTask(looping_task, true, true);

    LOG_TRACE(log, "Cleaning up RabbitMQ after table usage");

    /// Just a paranoid try catch, it is not actually needed.
    try
    {
        for (auto & consumer : consumers_ref)
            consumer.lock()->closeConnections();

        if (drop_table)
            cleanupRabbitMQ();

        /// It is important to close connection here - before removing consumers, because
        /// it will finish and clean callbacks, which might use those consumers data.
        connection->disconnect();

        for (size_t i = 0; i < num_created_consumers; ++i)
            popConsumer();
    }
    catch (...)
    {
        tryLogCurrentException(log);
    }

    StreamingStorageRegistry::instance().unregisterTable(getStorageID(), /* if_exists */ true);
    LOG_TRACE(log, "Shutdown finished");
}

void StorageRabbitMQ::renameInMemory(const StorageID & new_table_id)
{
    const auto prev_storage_id = getStorageID();
    IStorage::renameInMemory(new_table_id);
    StreamingStorageRegistry::instance().renameTable(prev_storage_id, getStorageID());
}


/// The only thing publishers are supposed to be aware of is _exchanges_ and queues are a responsibility of a consumer.
/// Therefore, if a table is dropped, a clean up is needed.