Merge pull request ClickHouse#88835 from ClickHouse/backport/25.8/88814

Backport ClickHouse#88814 to 25.8: Catch exceptions when async logging fails to prevent program aborts

Author: Algunenano

src/Loggers/OwnSplitChannel.cpp

@@ -487,20 +487,40 @@ void OwnAsyncSplitChannel::runChannel(size_t i)
 
     while (is_open)
     {
-        log_notification(notification);
-        notification = queues[i]->waitDequeueMessage();
+        try
+        {
+            log_notification(notification);
+            notification = queues[i]->waitDequeueMessage();
+        }
+        catch (...)
+        {
+            const std::string & exception_message = getCurrentExceptionMessage(true);
+            writeRetry(STDERR_FILENO, "Cannot log message in OwnAsyncSplitChannel channel: ");
+            writeRetry(STDERR_FILENO, exception_message.data(), exception_message.size());
+            writeRetry(STDERR_FILENO, "\n");
+        }
     }
 
-    /// Flush everything before closing
-    log_notification(notification);
-
-    /// We want to process only what's currently in the queue and not block other logging
-    auto queue = queues[i]->getCurrentQueueAndClear();
-    while (!queue.empty())
+    try
     {
-        notification = queue.front();
-        queue.pop_front();
+        /// Flush everything before closing
         log_notification(notification);
+
+        /// We want to process only what's currently in the queue and not block other logging
+        auto queue = queues[i]->getCurrentQueueAndClear();
+        while (!queue.empty())
+        {
+            notification = queue.front();
+            queue.pop_front();
+            log_notification(notification);
+        }
+    }
+    catch (...)
+    {
+        const std::string & exception_message = getCurrentExceptionMessage(true);
+        writeRetry(STDERR_FILENO, "Cannot flush messages in OwnAsyncSplitChannel channel: ");
+        writeRetry(STDERR_FILENO, exception_message.data(), exception_message.size());
+        writeRetry(STDERR_FILENO, "\n");
     }
 }
 
@@ -530,40 +550,59 @@ void OwnAsyncSplitChannel::runTextLog()
     auto notification = text_log_queue.waitDequeueMessage();
     while (is_open)
     {
-        if (flush_text_logs)
+        try
         {
-            auto text_log_locked = text_log.lock();
-            if (!text_log_locked)
-                return;
+            if (flush_text_logs)
+            {
+                auto text_log_locked = text_log.lock();
+                if (!text_log_locked)
+                    return;
 
-            if (notification)
-                log_notification(notification, text_log_locked);
+                if (notification)
+                    log_notification(notification, text_log_locked);
 
-            flush_queue(text_log_locked);
+                flush_queue(text_log_locked);
 
-            flush_text_logs = false;
-            flush_text_logs.notify_all();
+                flush_text_logs = false;
+                flush_text_logs.notify_all();
+            }
+            else if (notification)
+            {
+                auto text_log_locked = text_log.lock();
+                if (!text_log_locked)
+                    return;
+                log_notification(notification, text_log_locked);
+            }
+
+            notification = text_log_queue.waitDequeueMessage();
         }
-        else if (notification)
+        catch (...)
         {
-            auto text_log_locked = text_log.lock();
-            if (!text_log_locked)
-                return;
-            log_notification(notification, text_log_locked);
+            const std::string & exception_message = getCurrentExceptionMessage(true);
+            writeRetry(STDERR_FILENO, "Cannot log message in OwnAsyncSplitChannel text log: ");
+            writeRetry(STDERR_FILENO, exception_message.data(), exception_message.size());
+            writeRetry(STDERR_FILENO, "\n");
         }
-
-        notification = text_log_queue.waitDequeueMessage();
     }
 
-    /// We want to flush everything already in the queue before closing so all messages are logged
-    auto text_log_locked = text_log.lock();
-    if (!text_log_locked)
-        return;
-
-    if (notification)
-        log_notification(notification, text_log_locked);
+    try
+    {
+        /// We want to flush everything already in the queue before closing so all messages are logged
+        auto text_log_locked = text_log.lock();
+        if (!text_log_locked)
+            return;
 
-    flush_queue(text_log_locked);
+        if (notification)
+            log_notification(notification, text_log_locked);
+        flush_queue(text_log_locked);
+    }
+    catch (...)
+    {
+        const std::string & exception_message = getCurrentExceptionMessage(true);
+        writeRetry(STDERR_FILENO, "Cannot flush queue in OwnAsyncSplitChannel text log: ");
+        writeRetry(STDERR_FILENO, exception_message.data(), exception_message.size());
+        writeRetry(STDERR_FILENO, "\n");
+    }
 }
 
 void OwnAsyncSplitChannel::setChannelProperty(const std::string & channel_name, const std::string & name, const std::string & value)