diff --git a/quickwit/quickwit-rest-client/src/rest_client.rs b/quickwit/quickwit-rest-client/src/rest_client.rs index ea5eda5a391..da5be00cb05 100644 --- a/quickwit/quickwit-rest-client/src/rest_client.rs +++ b/quickwit/quickwit-rest-client/src/rest_client.rs @@ -343,18 +343,23 @@ impl QuickwitClient { let mut cumulated_resp = RestIngestResponse::default(); while let Some(batch) = batch_reader.next_batch().await? { loop { - let timeout = if !batch_reader.has_next() && last_block_commit != CommitType::Auto { + // Only the last batch carries the requested commit mode; the other batches + // use the default (auto) commit. + let is_last_batch = !batch_reader.has_next(); + let timeout = if is_last_batch && last_block_commit != CommitType::Auto { self.commit_timeout } else { self.ingest_timeout }; - match last_block_commit { - CommitType::Auto => {} - CommitType::WaitFor => { - query_params.insert("commit", "wait_for"); - } - CommitType::Force => { - query_params.insert("commit", "force"); + if is_last_batch { + match last_block_commit { + CommitType::Auto => {} + CommitType::WaitFor => { + query_params.insert("commit", "wait_for"); + } + CommitType::Force => { + query_params.insert("commit", "force"); + } } } let response = self @@ -961,6 +966,44 @@ mod test { assert_eq!(actual_response, mock_response); } + #[tokio::test] + async fn test_ingest_endpoint_force_commit_only_on_last_batch() { + let mock_server = MockServer::start().await; + let server_url = Url::parse(&mock_server.uri()).unwrap(); + let qw_client = QuickwitClientBuilder::new(server_url).build(); + let mock_response = RestIngestResponse { + num_docs_for_processing: 1, + num_ingested_docs: Some(1), + num_rejected_docs: Some(0), + parse_failures: Some(Vec::new()), + }; + Mock::given(method("POST")) + .and(path("/api/v1/my-index/ingest")) + .respond_with(ResponseTemplate::new(StatusCode::OK).set_body_json(&mock_response)) + .mount(&mock_server) + .await; + // Each document is 10 bytes long, so a 12-byte limit yields one batch per document. + let ingest_source = + IngestSource::Str("{\"id\": 1}\n{\"id\": 2}\n{\"id\": 3}\n".to_string()); + qw_client + .ingest("my-index", ingest_source, Some(12), None, CommitType::Force) + .await + .unwrap(); + let requests = mock_server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 3); + let commit_params: Vec> = requests + .iter() + .map(|request| { + request + .url + .query_pairs() + .find(|(key, _)| key == "commit") + .map(|(_, value)| value.to_string()) + }) + .collect(); + assert_eq!(commit_params, vec![None, None, Some("force".to_string())]); + } + #[tokio::test] async fn test_ingest_endpoint_with_wait_for_commit() { let mock_server = MockServer::start().await;