Flush messages for 1 sec before terminating kafka sink if we have in …

…flight messages.

Signed-off-by: Matthias Wahl <mwahl@wayfair.com>

src/sink/kafka.rs

@@ -30,7 +30,7 @@ use rdkafka::{
     error::KafkaError,
     producer::{FutureProducer, FutureRecord},
 };
-use std::fmt;
+use std::{fmt, time::Duration};
 
 #[derive(Deserialize)]
 pub struct Config {
@@ -314,4 +314,16 @@ impl Sink for Kafka {
     fn auto_ack(&self) -> bool {
         false
     }
+    async fn terminate(&mut self) {
+        if self.producer.in_flight_count() > 0 {
+            // wait a second in order to flush messages.
+            let wait_secs = 1;
+            info!(
+                "[Sink::{}] Flushing messages. Waiting for {} seconds.",
+                wait_secs, &self.sink_url
+            );
+            self.producer.flush(Duration::from_secs(1));
+            info!("[Sink::{}] Terminating.", &self.sink_url);
+        }
+    }
 }