refactor(connectors): use try_lock for restart guard to avoid queuing

Signed-off-by: seokjin0414 <sars21@hanmail.net>
Signed-off-by: shin <sars21@hanmail.net>

Author: seokjin0414

core/connectors/runtime/src/manager/sink.rs

@@ -249,7 +249,10 @@ impl SinkManager {
             let details = details.lock().await;
             details.restart_guard.clone()
         };
-        let _lock = guard.lock().await;
+        let Ok(_lock) = guard.try_lock() else {
+            info!("Restart already in progress for sink connector: {key}, skipping.");
+            return Ok(());
+        };
 
         info!("Restarting sink connector: {key}");
         self.stop_connector(key, metrics).await?;

core/connectors/runtime/src/manager/source.rs

@@ -257,7 +257,10 @@ impl SourceManager {
             let details = details.lock().await;
             details.restart_guard.clone()
         };
-        let _lock = guard.lock().await;
+        let Ok(_lock) = guard.try_lock() else {
+            info!("Restart already in progress for source connector: {key}, skipping.");
+            return Ok(());
+        };
 
         info!("Restarting source connector: {key}");
         self.stop_connector(key, metrics).await?;