fix(codecs): Improve deserialization of EncodingConfigAdapter / fix overriding framing (#12750)

* Improve `EncodingConfigAdapter` deserialization
* Add note about generic error message of untagged enum EncodingConfig

Signed-off-by: Pablo Sichert <[email protected]>

Author: pablosichert

lib/codecs/src/encoding/framing/character_delimited.rs

@@ -26,7 +26,7 @@ impl CharacterDelimitedEncoderConfig {
 }
 
 /// Options for building a `CharacterDelimitedEncoder`.
-#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
 pub struct CharacterDelimitedEncoderOptions {
     /// The character that delimits byte sequences.
     #[serde(with = "vector_core::serde::ascii_char")]

lib/codecs/src/encoding/mod.rs

@@ -49,7 +49,7 @@ impl From<std::io::Error> for Error {
 // Unfortunately, copying options of the nested enum variants is necessary
 // since `serde` doesn't allow `flatten`ing these:
 // https://github.com/serde-rs/serde/issues/1402.
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
 #[serde(tag = "method", rename_all = "snake_case")]
 pub enum FramingConfig {
     /// Configures the `BytesEncoder`.

src/sinks/aws_cloudwatch_logs/config.rs

@@ -80,12 +80,12 @@ impl ClientBuilder for CloudwatchLogsClientBuilder {
 }
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct CloudwatchLogsSinkConfig {
     pub group_name: Template,
     pub stream_name: Template,
     #[serde(flatten)]
     pub region: RegionOrEndpoint,
-    #[serde(flatten)]
     pub encoding:
         EncodingConfigAdapter<EncodingConfig<StandardEncodings>, StandardEncodingsMigrator>,
     pub create_missing_group: Option<bool>,

src/sinks/aws_kinesis_firehose/config.rs

@@ -51,11 +51,11 @@ impl SinkBatchSettings for KinesisFirehoseDefaultBatchSettings {
 }
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct KinesisFirehoseSinkConfig {
     pub stream_name: String,
     #[serde(flatten)]
     pub region: RegionOrEndpoint,
-    #[serde(flatten)]
     pub encoding:
         EncodingConfigAdapter<EncodingConfig<StandardEncodings>, StandardEncodingsMigrator>,
     #[serde(default)]

src/sinks/aws_kinesis_streams/config.rs

@@ -104,12 +104,12 @@ impl SinkBatchSettings for KinesisDefaultBatchSettings {
 }
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct KinesisSinkConfig {
     pub stream_name: String,
     pub partition_key_field: Option<String>,
     #[serde(flatten)]
     pub region: RegionOrEndpoint,
-    #[serde(flatten)]
     pub encoding:
         EncodingConfigAdapter<EncodingConfig<StandardEncodings>, StandardEncodingsMigrator>,
     #[serde(default)]

src/sinks/aws_s3/config.rs

@@ -36,6 +36,7 @@ const DEFAULT_FILENAME_TIME_FORMAT: &str = "%s";
 const DEFAULT_FILENAME_APPEND_UUID: bool = true;
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct S3SinkConfig {
     pub bucket: String,
     pub key_prefix: Option<String>,

src/sinks/aws_sqs/config.rs

@@ -46,11 +46,11 @@ impl EncodingConfigMigrator for EncodingMigrator {
 }
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct SqsSinkConfig {
     pub queue_url: String,
     #[serde(flatten)]
     pub region: RegionOrEndpoint,
-    #[serde(flatten)]
     pub encoding: EncodingConfigAdapter<EncodingConfig<Encoding>, EncodingMigrator>,
     pub message_group_id: Option<String>,
     pub message_deduplication_id: Option<String>,

src/sinks/azure_blob/config.rs

@@ -31,6 +31,7 @@ use crate::{
 };
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct AzureBlobSinkConfig {
     pub connection_string: String,
     pub(super) container_name: String,

src/sinks/file/mod.rs

@@ -61,6 +61,7 @@ impl EncodingConfigWithFramingMigrator for EncodingMigrator {
 }
 
 #[derive(Deserialize, Serialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct FileSinkConfig {
     pub path: Template,
     pub idle_timeout_secs: Option<u64>,

src/sinks/gcp/cloud_storage.rs

@@ -62,6 +62,7 @@ pub enum GcsHealthcheckError {
 const NAME: &str = "gcp_cloud_storage";
 
 #[derive(Deserialize, Serialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct GcsSinkConfig {
     bucket: String,
     acl: Option<GcsPredefinedAcl>,