enhancement(console sink): Add end-to-end acknowledgement support

src/sinks/console/config.rs

@@ -38,13 +38,20 @@ pub struct ConsoleSinkConfig {
         EncodingConfig<StandardEncodings>,
         StandardEncodingsWithFramingMigrator,
     >,
+    #[serde(
+        default,
+        deserialize_with = "crate::serde::bool_or_struct",
+        skip_serializing_if = "crate::serde::skip_serializing_if_default"
+    )]
+    pub acknowledgements: AcknowledgementsConfig,
 }
 
 impl GenerateConfig for ConsoleSinkConfig {
     fn generate_config() -> toml::Value {
         toml::Value::try_from(Self {
             target: Target::Stdout,
             encoding: EncodingConfig::from(StandardEncodings::Json).into(),
+            acknowledgements: Default::default(),
         })
         .unwrap()
     }
@@ -97,7 +104,7 @@ impl SinkConfig for ConsoleSinkConfig {
     }
 
     fn acknowledgements(&self) -> Option<&AcknowledgementsConfig> {
-        None
+        Some(&self.acknowledgements)
     }
 }
 

src/sinks/console/sink.rs

@@ -12,7 +12,7 @@ use vector_core::{
 
 use crate::{
     codecs::Encoder,
-    event::Event,
+    event::{Event, EventStatus, Finalizable},
     sinks::util::{encoding::Transformer, StreamSink},
 };
 
@@ -33,28 +33,36 @@ where
             let event_byte_size = event.size_of();
             self.transformer.transform(&mut event);
 
+            let finalizers = event.take_finalizers();
             let mut bytes = BytesMut::new();
             self.encoder.encode(event, &mut bytes).map_err(|_| {
                 // Error is handled by `Encoder`.
+                finalizers.update_status(EventStatus::Errored);
             })?;
 
-            self.output.write_all(&bytes).await.map_err(|error| {
-                // Error when writing to stdout/stderr is likely irrecoverable,
-                // so stop the sink.
-                error!(message = "Error writing to output. Stopping sink.", %error);
-            })?;
-
-            self.acker.ack(1);
-
-            emit!(EventsSent {
-                byte_size: event_byte_size,
-                count: 1,
-                output: None,
-            });
-            emit!(BytesSent {
-                byte_size: bytes.len(),
-                protocol: "console"
-            });
+            match self.output.write_all(&bytes).await {
+                Err(error) => {
+                    // Error when writing to stdout/stderr is likely irrecoverable,
+                    // so stop the sink.
+                    error!(message = "Error writing to output. Stopping sink.", %error);
+                    finalizers.update_status(EventStatus::Errored);
+                    return Err(());
+                }
+                Ok(()) => {
+                    finalizers.update_status(EventStatus::Delivered);
+                    self.acker.ack(1);
+
+                    emit!(EventsSent {
+                        byte_size: event_byte_size,
+                        count: 1,
+                        output: None,
+                    });
+                    emit!(BytesSent {
+                        byte_size: bytes.len(),
+                        protocol: "console"
+                    });
+                }
+            }
         }
 
         Ok(())

src/topology/test/doesnt_reload.rs

@@ -20,6 +20,7 @@ async fn topology_doesnt_reload_new_data_dir() {
         ConsoleSinkConfig {
             target: Target::Stdout,
             encoding: EncodingConfig::from(StandardEncodings::Text).into(),
+            acknowledgements: Default::default(),
         },
     );
     old_config.global.data_dir = Some(Path::new("/asdf").to_path_buf());

src/topology/test/reload.rs

@@ -39,6 +39,7 @@ async fn topology_reuse_old_port() {
         ConsoleSinkConfig {
             target: Target::Stdout,
             encoding: EncodingConfig::from(StandardEncodings::Text).into(),
+            acknowledgements: Default::default(),
         },
     );
 
@@ -50,6 +51,7 @@ async fn topology_reuse_old_port() {
         ConsoleSinkConfig {
             target: Target::Stdout,
             encoding: EncodingConfig::from(StandardEncodings::Text).into(),
+            acknowledgements: Default::default(),
         },
     );
 
@@ -73,6 +75,7 @@ async fn topology_rebuild_old() {
         ConsoleSinkConfig {
             target: Target::Stdout,
             encoding: EncodingConfig::from(StandardEncodings::Text).into(),
+            acknowledgements: Default::default(),
         },
     );
 
@@ -84,6 +87,7 @@ async fn topology_rebuild_old() {
         ConsoleSinkConfig {
             target: Target::Stdout,
             encoding: EncodingConfig::from(StandardEncodings::Text).into(),
+            acknowledgements: Default::default(),
         },
     );
 
@@ -109,6 +113,7 @@ async fn topology_old() {
         ConsoleSinkConfig {
             target: Target::Stdout,
             encoding: EncodingConfig::from(StandardEncodings::Text).into(),
+            acknowledgements: Default::default(),
         },
     );
 

src/topology/test/source_finished.rs

@@ -21,6 +21,7 @@ async fn sources_finished() {
         ConsoleSinkConfig {
             target: Target::Stdout,
             encoding: EncodingConfig::from(StandardEncodings::Text).into(),
+            acknowledgements: Default::default(),
         },
     );
 

website/cue/reference/components/sinks/console.cue

@@ -13,7 +13,7 @@ components: sinks: console: {
 	}
 
 	features: {
-		acknowledgements: false
+		acknowledgements: true
 		healthcheck: enabled: false
 		send: {
 			compression: enabled: false