diff --git a/src/servers/src/pending_rows_batcher.rs b/src/servers/src/pending_rows_batcher.rs
index d0eb1616090..e0494a28f11 100644
--- a/src/servers/src/pending_rows_batcher.rs
+++ b/src/servers/src/pending_rows_batcher.rs
@@ -2151,13 +2151,14 @@ mod tests {
     use tokio::time::{advance, sleep};
 
     use super::{
-        BatchKey, Error, FlushBatch, FlushRegionWrite, FlushWaiter, MetricRowBatchProtocol,
-        PendingBatch, PendingRowsBatchMode, PendingRowsBatcher, PendingRowsSchemaAlterer,
-        PendingWorker, PhysicalFlushCatalogProvider, PhysicalFlushNodeRequester,
-        PhysicalFlushPartitionProvider, PhysicalTableMetadata, PlannedRegionBatch,
-        RecordBatchWithTsIdx, ResolvedRegionBatch, TableBatch, WorkerCommand, columns_taxonomy,
-        drain_batch, encode_region_write_requests, extract_timestamps, flush_batch,
-        flush_batch_physical, flush_region_writes_concurrently, greptime_timestamp,
+        BatchKey, CatalogManagerPhysicalFlushAdapter, Error, FlushBatch, FlushRegionWrite,
+        FlushWaiter, MetricRowBatchProtocol, NodeManagerPhysicalFlushAdapter,
+        PartitionManagerPhysicalFlushAdapter, PendingBatch, PendingRowsBatchMode,
+        PendingRowsBatcher, PendingRowsSchemaAlterer, PendingWorker, PhysicalFlushCatalogProvider,
+        PhysicalFlushNodeRequester, PhysicalFlushPartitionProvider, PhysicalTableMetadata,
+        PlannedRegionBatch, RecordBatchWithTsIdx, ResolvedRegionBatch, TableBatch, WorkerCommand,
+        columns_taxonomy, drain_batch, encode_region_write_requests, extract_timestamps,
+        flush_batch, flush_batch_physical, flush_region_writes_concurrently, greptime_timestamp,
         is_scalar_metric_batchable, notify_flow_dirty_windows_after_flush, plan_region_batches,
         remove_worker_if_same_channel, should_close_worker_on_idle_timeout,
         should_dispatch_concurrently, start_flow_notification_worker, start_worker,
@@ -3841,10 +3842,10 @@ mod tests {
             worker_tx.clone(),
             workers.clone(),
             worker_rx,
-            shutdown.clone(),
-            partition_manager,
-            node_manager,
-            catalog_manager,
+            shutdown.subscribe(),
+            Arc::new(PartitionManagerPhysicalFlushAdapter { partition_manager }),
+            Arc::new(NodeManagerPhysicalFlushAdapter { node_manager }),
+            Arc::new(CatalogManagerPhysicalFlushAdapter { catalog_manager }),
             flow_notification_tx,
             flush_interval,
             worker_idle_timeout,