[Vector] Disable content encoding for GCP sink

src/sinks/gcp/cloud_storage.rs

@@ -147,6 +147,11 @@ pub struct GcsSinkConfig {
     #[serde(flatten)]
     encoding: EncodingConfigWithFraming,
 
+    /// Whether to set content encoding for the object
+    #[serde(default = "crate::serde::default_true")]
+    #[configurable(metadata(docs::advanced))]
+    set_content_encoding: bool,
+
     #[configurable(derived)]
     #[serde(default)]
     compression: Compression,
@@ -194,6 +199,7 @@ fn default_config(encoding: EncodingConfigWithFraming) -> GcsSinkConfig {
         filename_append_uuid: true,
         filename_extension: Default::default(),
         encoding,
+        set_content_encoding: true,
         compression: Compression::gzip_default(),
         batch: Default::default(),
         request: Default::default(),
@@ -377,10 +383,14 @@ impl RequestSettings {
             .acl
             .map(|acl| HeaderValue::from_str(&to_string(acl)).unwrap());
         let content_type = HeaderValue::from_str(encoder.content_type()).unwrap();
-        let content_encoding = config
-            .compression
-            .content_encoding()
-            .map(|ce| HeaderValue::from_str(&to_string(ce)).unwrap());
+        let content_encoding = if config.set_content_encoding {
+            config
+                .compression
+                .content_encoding()
+                .map(|ce| HeaderValue::from_str(&to_string(ce)).unwrap())
+        } else {
+            None
+        };
         let storage_class = config.storage_class.unwrap_or_default();
         let storage_class = HeaderValue::from_str(&to_string(storage_class)).unwrap();
         let metadata = config
@@ -502,13 +512,19 @@ mod tests {
         RequestSettings::new(sink_config, context).expect("Could not create request settings")
     }
 
-    fn build_request(extension: Option<&str>, uuid: bool, compression: Compression) -> GcsRequest {
+    fn build_request(
+        extension: Option<&str>,
+        uuid: bool,
+        compression: Compression,
+        set_content_encoding: bool,
+    ) -> GcsRequest {
         let context = SinkContext::default();
         let sink_config = GcsSinkConfig {
             key_prefix: Some("key/".into()),
             filename_time_format: "date".into(),
             filename_extension: extension.map(Into::into),
             filename_append_uuid: uuid,
+            set_content_encoding: set_content_encoding,
             compression,
             ..default_config(
                 (
@@ -539,16 +555,28 @@ mod tests {
 
     #[test]
     fn gcs_build_request() {
-        let req = build_request(Some("ext"), false, Compression::None);
+        let req = build_request(Some("ext"), false, Compression::None, true);
         assert_eq!(req.key, "key/date.ext".to_string());
 
-        let req = build_request(None, false, Compression::None);
+        let req = build_request(None, false, Compression::None, true);
         assert_eq!(req.key, "key/date.log".to_string());
 
-        let req = build_request(None, false, Compression::gzip_default());
+        let req = build_request(None, false, Compression::gzip_default(), true);
         assert_eq!(req.key, "key/date.log.gz".to_string());
 
-        let req = build_request(None, true, Compression::gzip_default());
+        let req = build_request(None, true, Compression::gzip_default(), true);
         assert_ne!(req.key, "key/date.log.gz".to_string());
     }
+
+    #[test]
+    fn gcs_build_request_set_content_encoding() {
+        let request = build_request(None, false, Compression::gzip_default(), true);
+        assert_eq!(
+            request.settings.content_encoding,
+            HeaderValue::from_str(Compression::gzip_default().content_encoding().unwrap()).ok()
+        );
+
+        let request = build_request(None, false, Compression::gzip_default(), false);
+        assert!(request.settings.content_encoding.is_none());
+    }
 }