feat(gcp_cloud_storage sink): add content_type option

changelog.d/gcp_cloud_storage_content_type.enhancement.md

@@ -0,0 +1,3 @@
+Add `content_type` option to the `gcp_cloud_storage` sink to override the `Content-Type` of created objects. If unset, defaults to the encoder's content type.
+
+authors: AnuragEkkati

src/sinks/gcp/cloud_storage.rs

@@ -161,6 +161,17 @@ pub struct GcsSinkConfig {
     #[serde(default)]
     compression: Compression,
 
+    /// Overrides the MIME type of the created objects.
+    ///
+    /// Directly comparable to the `Content-Type` HTTP header.
+    ///
+    /// If not specified, defaults to the encoder's content type.
+    #[configurable(metadata(
+        docs::examples = "text/plain; charset=utf-8",
+        docs::examples = "application/gzip"
+    ))]
+    content_type: Option<String>,
+
     #[configurable(derived)]
     #[serde(default)]
     batch: BatchConfig<BulkSizeBasedDefaultBatchSettings>,
@@ -209,6 +220,7 @@ fn default_config(encoding: EncodingConfigWithFraming) -> GcsSinkConfig {
         filename_time_format: default_time_format(),
         filename_append_uuid: true,
         filename_extension: Default::default(),
+        content_type: Default::default(),
         encoding,
         compression: Compression::gzip_default(),
         batch: Default::default(),
@@ -394,7 +406,11 @@ impl RequestSettings {
         let acl = config
             .acl
             .map(|acl| HeaderValue::from_str(&to_string(acl)).unwrap());
-        let content_type = HeaderValue::from_str(encoder.content_type()).unwrap();
+        let content_type_str = config
+            .content_type
+            .as_deref()
+            .unwrap_or_else(|| encoder.content_type());
+        let content_type = HeaderValue::from_str(content_type_str)?;
         let content_encoding = config
             .compression
             .content_encoding()
@@ -572,4 +588,50 @@ mod tests {
         let req = build_request(None, true, Compression::gzip_default());
         assert_ne!(req.key, "key/date.log.gz".to_string());
     }
+
+    #[test]
+    fn gcs_content_type_default() {
+        let context = SinkContext::default();
+        let sink_config = GcsSinkConfig {
+            content_type: None,
+            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
+        };
+
+        let request_settings = request_settings(&sink_config, context);
+        // Should default to encoder's content type which is "text/plain" for text codec
+        assert_eq!(
+            request_settings.content_type.to_str().unwrap(),
+            "text/plain"
+        );
+    }
+
+    #[test]
+    fn gcs_content_type_custom() {
+        let context = SinkContext::default();
+        let sink_config = GcsSinkConfig {
+            content_type: Some("text/plain; charset=utf-8".to_string()),
+            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
+        };
+
+        let request_settings = request_settings(&sink_config, context);
+        // Should use custom content type
+        assert_eq!(
+            request_settings.content_type.to_str().unwrap(),
+            "text/plain; charset=utf-8"
+        );
+    }
+
+    #[test]
+    fn gcs_content_type_invalid() {
+        let context = SinkContext::default();
+        let sink_config = GcsSinkConfig {
+            // Invalid header value with newline character
+            content_type: Some("text/plain\nInvalid".to_string()),
+            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
+        };
+
+        let result = RequestSettings::new(&sink_config, context);
+        // Should return an error, not panic
+        assert!(result.is_err());
+    }
 }

website/cue/reference/components/sinks/generated/gcp_cloud_storage.cue

@@ -172,6 +172,17 @@ generated: components: sinks: gcp_cloud_storage: configuration: {
 			}
 		}
 	}
+	content_type: {
+		description: """
+			Overrides the MIME type of the created objects.
+
+			Directly comparable to the `Content-Type` HTTP header.
+
+			If not specified, defaults to the encoder's content type.
+			"""
+		required: false
+		type: string: examples: ["text/plain; charset=utf-8", "application/gzip"]
+	}
 	credentials_path: {
 		description: """
 			Path to a [service account][gcp_service_account_credentials] credentials JSON file.