fix(compactor): Fix an unexpected termination of shared compactor (ri…

…singwavelabs#19505) (risingwavelabs#19517)

Co-authored-by: Li0k <[email protected]>

src/storage/compactor/src/server.rs

@@ -363,7 +363,9 @@ pub async fn shared_compactor_serve(
         await_tree_reg,
     };
 
-    risingwave_storage::hummock::compactor::start_shared_compactor(
+    // TODO(shutdown): don't collect there's no need to gracefully shutdown them.
+    // Hold the join handle and tx to keep the compactor running.
+    let _compactor_handle = risingwave_storage::hummock::compactor::start_shared_compactor(
         grpc_proxy_client,
         receiver,
         compactor_context,

src/storage/src/hummock/compactor/mod.rs

@@ -275,6 +275,7 @@ impl Compactor {
 /// The background compaction thread that receives compaction tasks from hummock compaction
 /// manager and runs compaction tasks.
 #[cfg_attr(coverage, coverage(off))]
+#[must_use]
 pub fn start_compactor(
     compactor_context: CompactorContext,
     hummock_meta_client: Arc<dyn HummockMetaClient>,
@@ -608,6 +609,7 @@ pub fn start_compactor(
 /// The background compaction thread that receives compaction tasks from hummock compaction
 /// manager and runs compaction tasks.
 #[cfg_attr(coverage, coverage(off))]
+#[must_use]
 pub fn start_shared_compactor(
     grpc_proxy_client: GrpcCompactorProxyClient,
     mut receiver: mpsc::UnboundedReceiver<Request<DispatchCompactionTaskRequest>>,