wait for merge at end of local ingest (#5542)

quickwit-oss · Nov 18, 2024 · fe0f39a · fe0f39a
1 parent 23289a1
commit fe0f39a
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 4 deletions.
diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs
@@ -495,7 +495,11 @@ pub async fn local_ingest_docs_cli(args: LocalIngestDocsArgs) -> anyhow::Result<
     let statistics =
         start_statistics_reporting_loop(indexing_pipeline_handle, args.input_path_opt.is_none())
             .await?;
-    merge_pipeline_handle.quit().await;
+    merge_pipeline_handle
+        .mailbox()
+        .ask(quickwit_indexing::FinishPendingMergesAndShutdownPipeline)
+        .await?;
+    merge_pipeline_handle.join().await;
     // Shutdown the indexing server.
     universe
         .send_exit_with_success(&indexing_server_mailbox)

diff --git a/quickwit/quickwit-indexing/src/actors/mod.rs b/quickwit/quickwit-indexing/src/actors/mod.rs
@@ -41,7 +41,7 @@ pub use indexer::{Indexer, IndexerCounters};
 pub use indexing_pipeline::{IndexingPipeline, IndexingPipelineParams};
 pub use indexing_service::{IndexingService, IndexingServiceCounters, INDEXING_DIR_NAME};
 pub use merge_executor::{combine_partition_ids, merge_split_attrs, MergeExecutor};
-pub use merge_pipeline::MergePipeline;
+pub use merge_pipeline::{FinishPendingMergesAndShutdownPipeline, MergePipeline};
 pub(crate) use merge_planner::{MergePlanner, RunFinalizeMergePolicyAndQuit};
 pub use merge_scheduler_service::{schedule_merge, MergePermit, MergeSchedulerService};
 pub use merge_split_downloader::MergeSplitDownloader;

diff --git a/quickwit/quickwit-indexing/src/lib.rs b/quickwit/quickwit-indexing/src/lib.rs
@@ -31,8 +31,8 @@ use tracing::info;
 
 use crate::actors::MergeSchedulerService;
 pub use crate::actors::{
-    IndexingError, IndexingPipeline, IndexingPipelineParams, IndexingService, PublisherType,
-    Sequencer, SplitsUpdateMailbox,
+    FinishPendingMergesAndShutdownPipeline, IndexingError, IndexingPipeline,
+    IndexingPipelineParams, IndexingService, PublisherType, Sequencer, SplitsUpdateMailbox,
 };
 pub use crate::controlled_directory::ControlledDirectory;
 use crate::models::IndexingStatistics;