add dead letter fallback

Author: ArunPiduguDD

src/sinks/aws_s3/config.rs

@@ -224,7 +224,7 @@ impl S3SinkConfig {
             .map(|ssekms_key_id| Template::try_from(ssekms_key_id.as_str()))
             .transpose()?;
 
-        let partitioner = S3KeyPartitioner::new(key_prefix, ssekms_key_id);
+        let partitioner = S3KeyPartitioner::new(key_prefix, ssekms_key_id, None);
 
         let transformer = self.encoding.transformer();
         let (framer, serializer) = self.encoding.build(SinkType::MessageBased)?;

src/sinks/azure_blob/config.rs

@@ -264,6 +264,6 @@ impl AzureBlobSinkConfig {
     }
 
     pub fn key_partitioner(&self) -> crate::Result<KeyPartitioner> {
-        Ok(KeyPartitioner::new(self.blob_prefix.clone()))
+        Ok(KeyPartitioner::new(self.blob_prefix.clone(), None))
     }
 }

src/sinks/gcp/cloud_storage.rs

@@ -284,6 +284,7 @@ impl GcsSinkConfig {
         Ok(KeyPartitioner::new(
             Template::try_from(self.key_prefix.as_deref().unwrap_or("date=%F/"))
                 .context(KeyPrefixTemplateSnafu)?,
+            None,
         ))
     }
 }

src/sinks/s3_common/partitioner.rs

@@ -9,14 +9,19 @@ pub struct S3PartitionKey {
 }
 
 /// Partitions items based on the generated key for the given event.
-pub struct S3KeyPartitioner(Template, Option<Template>);
+pub struct S3KeyPartitioner(Template, Option<Template>, Option<String>);
 
 impl S3KeyPartitioner {
     pub const fn new(
         key_prefix_template: Template,
         ssekms_key_id_template: Option<Template>,
+        dead_letter_key_prefix: Option<String>,
     ) -> Self {
-        Self(key_prefix_template, ssekms_key_id_template)
+        Self(
+            key_prefix_template,
+            ssekms_key_id_template,
+            dead_letter_key_prefix,
+        )
     }
 }
 
@@ -28,14 +33,24 @@ impl Partitioner for S3KeyPartitioner {
         let key_prefix = self
             .0
             .render_string(item)
-            .map_err(|error| {
-                emit!(TemplateRenderingError {
-                    error,
-                    field: Some("key_prefix"),
-                    drop_event: true,
-                });
+            .or_else(|error| {
+                if let Some(dead_letter_key_prefix) = &self.2 {
+                    emit!(TemplateRenderingError {
+                        error,
+                        field: Some("key_prefix"),
+                        drop_event: false,
+                    });
+                    Ok(dead_letter_key_prefix.clone())
+                } else {
+                    Err(emit!(TemplateRenderingError {
+                        error,
+                        field: Some("key_prefix"),
+                        drop_event: true,
+                    }))
+                }
             })
             .ok()?;
+
         let ssekms_key_id = self
             .1
             .as_ref()

src/sinks/util/partitioner.rs

@@ -3,11 +3,11 @@ use vector_lib::{event::Event, partition::Partitioner};
 use crate::{internal_events::TemplateRenderingError, template::Template};
 
 /// Partitions items based on the generated key for the given event.
-pub struct KeyPartitioner(Template);
+pub struct KeyPartitioner(Template, Option<String>);
 
 impl KeyPartitioner {
-    pub const fn new(template: Template) -> Self {
-        Self(template)
+    pub const fn new(template: Template, dead_letter_key_prefix: Option<String>) -> Self {
+        Self(template, dead_letter_key_prefix)
     }
 }
 
@@ -18,12 +18,21 @@ impl Partitioner for KeyPartitioner {
     fn partition(&self, item: &Self::Item) -> Self::Key {
         self.0
             .render_string(item)
-            .map_err(|error| {
-                emit!(TemplateRenderingError {
-                    error,
-                    field: Some("key_prefix"),
-                    drop_event: true,
-                });
+            .or_else(|error| {
+                if let Some(dead_letter_key_prefix) = &self.1 {
+                    emit!(TemplateRenderingError {
+                        error,
+                        field: Some("key_prefix"),
+                        drop_event: false,
+                    });
+                    Ok(dead_letter_key_prefix.clone())
+                } else {
+                    Err(emit!(TemplateRenderingError {
+                        error,
+                        field: Some("key_prefix"),
+                        drop_event: true,
+                    }))
+                }
             })
             .ok()
     }

src/sinks/webhdfs/config.rs

@@ -160,6 +160,6 @@ impl WebHdfsConfig {
 
     pub fn key_partitioner(&self) -> crate::Result<KeyPartitioner> {
         let prefix = self.prefix.clone().try_into()?;
-        Ok(KeyPartitioner::new(prefix))
+        Ok(KeyPartitioner::new(prefix, None))
     }
 }