From 0d86eabdcb2df93eaae1d86a7072916540f030a6 Mon Sep 17 00:00:00 2001
From: Ashish Kolhe <35160958+oasisk@users.noreply.github.com>
Date: Tue, 19 Mar 2024 12:22:33 +0530
Subject: [PATCH] fix: add debug for task panic & etcd lock error (#3040)

---
 src/job/files/parquet.rs | 20 ++++++++++++++++++--
 src/service/schema.rs    |  2 +-
 2 files changed, 19 insertions(+), 3 deletions(-)

diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs
index c48f379707..2937d7d5ee 100644
--- a/src/job/files/parquet.rs
+++ b/src/job/files/parquet.rs
@@ -15,6 +15,7 @@
 
 use std::{collections::HashMap, io::Cursor, path::Path, sync::Arc, time::UNIX_EPOCH};
 
+use anyhow::Context;
 use arrow::{
     array::{ArrayRef, BooleanArray, Int64Array, StringArray},
     record_batch::RecordBatch,
@@ -135,7 +136,11 @@ pub async fn move_files_to_storage() -> Result<(), anyhow::Error> {
         }
 
         let wal_dir = wal_dir.clone();
-        let permit = semaphore.clone().acquire_owned().await.unwrap();
+        let permit = semaphore
+            .clone()
+            .acquire_owned()
+            .await
+            .context("[INGESTER:JOB] Failed to acquire semaphore for file uploading")?;
         let task: JoinHandle<Result<(), anyhow::Error>> = tokio::task::spawn(async move {
             // sort by created time
             let mut files_with_size = files_with_size.to_owned();
@@ -175,7 +180,18 @@ pub async fn move_files_to_storage() -> Result<(), anyhow::Error> {
             }
 
             // get latest schema
-            let latest_schema = db::schema::get(&org_id, &stream_name, stream_type).await?;
+            let latest_schema = db::schema::get(&org_id, &stream_name, stream_type)
+                .await
+                .map_err(|e| {
+                    log::error!(
+                        "[INGESTER:JOB] Failed to get latest schema for stream [{}/{}/{}]: {}",
+                        &org_id,
+                        stream_type,
+                        &stream_name,
+                        e
+                    );
+                    e
+                })?;
 
             // start merge files and upload to s3
             loop {
diff --git a/src/service/schema.rs b/src/service/schema.rs
index d089ca795b..12d004cd74 100644
--- a/src/service/schema.rs
+++ b/src/service/schema.rs
@@ -596,7 +596,7 @@ async fn handle_diff_schema_cluster_mode(
     let locker = match dist_lock::lock(&lock_key, 0).await {
         Ok(v) => v,
         Err(e) => {
-            log::error!("dist_lock for handle diff schema error: {}", e);
+            log::error!("[SCHEMA] dist_lock schema {} error : {}", lock_key, e);
             return None;
         }
     };