@@ -257,39 +257,75 @@ pub fn build_headers(headers: &Vec<KafkaHeader>) -> OwnedHeaders { |
257 | 257 | result |
258 | 258 | } |
259 | 259 |
|
| 260 | +/// A produce call that librdkafka refused for a reason other than a full |
| 261 | +/// queue. The exactly-once sink reads the Kafka error to decide whether the |
| 262 | +/// open transaction must be aborted. |
| 263 | +#[derive(Debug)] |
| 264 | +pub(crate) struct SendError { |
| 265 | + pub topic: String, |
| 266 | + pub error: KafkaError, |
| 267 | +} |
| 268 | + |
| 269 | +impl std::fmt::Display for SendError { |
| 270 | + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 271 | + write!( |
| 272 | + f, |
| 273 | + "error sending Kafka message to topic '{}': {}", |
| 274 | + self.topic, self.error |
| 275 | + ) |
| 276 | + } |
| 277 | +} |
| 278 | + |
| 279 | +impl std::error::Error for SendError { |
| 280 | + // `Display` prints the Kafka error already. Returning it here too would |
| 281 | + // make `{:#}` print it twice. |
| 282 | + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { |
| 283 | + None |
| 284 | + } |
| 285 | +} |
| 286 | + |
260 | 287 | /// Hands `record` to `producer`, waiting out a full producer queue. |
261 | 288 | /// |
262 | 289 | /// The wait is unbounded, so it ends on `shutdown`: this runs on the |
263 | 290 | /// controller's output thread, a DBSP aux thread that `RuntimeHandle::kill` |
264 | 291 | /// joins, and a queue that a down broker keeps full would otherwise hold the |
265 | 292 | /// teardown open for as long as the broker stays down. |
| 293 | +/// |
| 294 | +/// Returns the time spent waiting, zero when the queue had room. `on_wait` |
| 295 | +/// receives the cumulative wait after each attempt, including the final |
| 296 | +/// attempt on success or failure, so metrics include interrupted waits. |
266 | 297 | pub fn kafka_send<T1, T2, C>( |
267 | 298 | producer: &ThreadedProducer<C>, |
268 | 299 | topic: &str, |
269 | 300 | mut record: BaseRecord<T1, T2, C::DeliveryOpaque>, |
270 | 301 | shutdown: &CancellationToken, |
271 | | -) -> AnyResult<()> |
| 302 | + on_wait: &mut dyn FnMut(Duration), |
| 303 | +) -> AnyResult<Duration> |
272 | 304 | where |
273 | 305 | T1: ToBytes + ?Sized, |
274 | 306 | T2: ToBytes + ?Sized, |
275 | 307 | C: ProducerContext, |
276 | 308 | { |
277 | 309 | let mut polling_interval = Duration::from_micros(10); |
278 | | - let mut start; |
| 310 | + let mut start: Option<Instant> = None; |
279 | 311 |
|
280 | 312 | loop { |
281 | | - match producer.send(record) { |
282 | | - Ok(()) => return Ok(()), |
| 313 | + let result = producer.send(record); |
| 314 | + if let Some(start) = start { |
| 315 | + on_wait(start.elapsed()); |
| 316 | + } |
| 317 | + match result { |
| 318 | + Ok(()) => return Ok(start.map(|start| start.elapsed()).unwrap_or_default()), |
283 | 319 | Err((e, r)) => match e { |
284 | 320 | KafkaError::MessageProduction(e) if is_retriable_send_error(e) => { |
285 | 321 | // Start timing after the first error. |
286 | | - start = Some(Instant::now()); |
| 322 | + let start = *start.get_or_insert_with(Instant::now); |
287 | 323 | record = r; |
288 | 324 |
|
289 | 325 | if shutdown.is_cancelled() { |
290 | 326 | bail!( |
291 | | - "gave up sending a message to Kafka topic '{topic}' after {:?} because the pipeline is shutting down (error: {e})", |
292 | | - start.unwrap().elapsed() |
| 327 | + "gave up sending a message to Kafka topic '{topic}' after {:?} because the pipeline is shutting down (error: {e})", |
| 328 | + start.elapsed() |
293 | 329 | ); |
294 | 330 | } |
295 | 331 |
|
@@ -298,7 +334,7 @@ where |
298 | 334 | warn!( |
299 | 335 | "Attempts to send a message to Kafka topic '{}' have failed for {:?}, will keep retrying (error: {e})", |
300 | 336 | &topic, |
301 | | - start.unwrap().elapsed() |
| 337 | + start.elapsed() |
302 | 338 | ); |
303 | 339 | } |
304 | 340 |
|
@@ -307,7 +343,13 @@ where |
307 | 343 | polling_interval = min(polling_interval * 2, MAX_POLLING_INTERVAL); |
308 | 344 | } |
309 | 345 | } |
310 | | - _ => bail!("error sending Kafka message: {e}"), |
| 346 | + _ => { |
| 347 | + return Err(SendError { |
| 348 | + topic: topic.to_owned(), |
| 349 | + error: e, |
| 350 | + } |
| 351 | + .into()); |
| 352 | + } |
311 | 353 | }, |
312 | 354 | } |
313 | 355 | } |
@@ -607,6 +649,7 @@ mod send_tests { |
607 | 649 | use super::{MAX_POLLING_INTERVAL, kafka_send}; |
608 | 650 | use rdkafka::ClientConfig; |
609 | 651 | use rdkafka::producer::{BaseRecord, DefaultProducerContext, ThreadedProducer}; |
| 652 | + use std::sync::atomic::{AtomicU64, Ordering}; |
610 | 653 | use std::sync::{Arc, mpsc}; |
611 | 654 | use std::time::Duration; |
612 | 655 | use tokio_util::sync::CancellationToken; |
@@ -635,6 +678,7 @@ mod send_tests { |
635 | 678 | "test_topic", |
636 | 679 | BaseRecord::<(), [u8], ()>::to("test_topic").payload(b"first".as_slice()), |
637 | 680 | &shutdown, |
| 681 | + &mut |_| {}, |
638 | 682 | ) |
639 | 683 | .expect("the first message must fit in the queue"); |
640 | 684 |
|
@@ -643,12 +687,17 @@ mod send_tests { |
643 | 687 | let sending_producer = producer.clone(); |
644 | 688 | let sending_shutdown = shutdown.clone(); |
645 | 689 | let (done, send_returned) = mpsc::channel(); |
| 690 | + let reported_wait = Arc::new(AtomicU64::new(0)); |
| 691 | + let sending_wait = reported_wait.clone(); |
646 | 692 | let sender = std::thread::spawn(move || { |
647 | 693 | let result = kafka_send( |
648 | 694 | &sending_producer, |
649 | 695 | "test_topic", |
650 | 696 | BaseRecord::<(), [u8], ()>::to("test_topic").payload(b"second".as_slice()), |
651 | 697 | &sending_shutdown, |
| 698 | + &mut |waited| { |
| 699 | + sending_wait.store(waited.as_micros() as u64, Ordering::Release); |
| 700 | + }, |
652 | 701 | ); |
653 | 702 | let _ = done.send(result.err()); |
654 | 703 | }); |
@@ -657,13 +706,21 @@ mod send_tests { |
657 | 706 | send_returned.recv_timeout(MAX_POLLING_INTERVAL).is_err(), |
658 | 707 | "a full queue must make `kafka_send` wait" |
659 | 708 | ); |
| 709 | + assert!( |
| 710 | + reported_wait.load(Ordering::Acquire) > 0, |
| 711 | + "report an active wait" |
| 712 | + ); |
660 | 713 |
|
661 | 714 | shutdown.cancel(); |
662 | 715 | let error = send_returned |
663 | 716 | .recv_timeout(Duration::from_secs(30)) |
664 | 717 | .expect("a shutdown must end the wait") |
665 | 718 | .expect("a send that gave up must report an error"); |
666 | 719 | sender.join().unwrap(); |
| 720 | + assert!( |
| 721 | + reported_wait.load(Ordering::Acquire) >= MAX_POLLING_INTERVAL.as_micros() as u64, |
| 722 | + "report the entire wait even when shutdown interrupts it" |
| 723 | + ); |
667 | 724 |
|
668 | 725 | assert!( |
669 | 726 | error.to_string().contains("shutting down"), |
|