feat(codecs): add 'use_json_names' options to protobuf codecs

changelog.d/protobuf_use_json_names.enhancement.md

@@ -0,0 +1,5 @@
+Added `use_json_names` option to protobuf encoding and decoding.
+When enabled, the codec uses JSON field names (camelCase) instead of protobuf field names (snake_case).
+This is useful when working with data that uses JSON naming conventions.
+
+authors: pront

lib/codecs/src/decoding/format/protobuf.rs

@@ -81,6 +81,16 @@ pub struct ProtobufDeserializerOptions {
     /// The name of the message type to use for serializing.
     #[configurable(metadata(docs::examples = "package.Message"))]
     pub message_type: String,
+
+    /// Use JSON field names (camelCase) instead of protobuf field names (snake_case).
+    ///
+    /// When enabled, the deserializer will output fields using their JSON names as defined
+    /// in the `.proto` file (e.g., `jobDescription` instead of `job_description`).
+    ///
+    /// This is useful when working with data that needs to be converted to JSON or
+    /// when interfacing with systems that use JSON naming conventions.
+    #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
+    pub use_json_names: bool,
 }
 
 /// Deserializer that builds `Event`s from a byte frame containing protobuf.
@@ -166,7 +176,12 @@ impl TryFrom<&ProtobufDeserializerConfig> for ProtobufDeserializer {
     fn try_from(config: &ProtobufDeserializerConfig) -> vector_common::Result<Self> {
         let message_descriptor =
             get_message_descriptor(&config.protobuf.desc_file, &config.protobuf.message_type)?;
-        Ok(Self::new(message_descriptor))
+        Ok(Self {
+            message_descriptor,
+            options: Options {
+                use_json_names: config.protobuf.use_json_names,
+            },
+        })
     }
 }
 

lib/codecs/src/encoding/format/protobuf.rs

@@ -30,7 +30,9 @@ impl ProtobufSerializerConfig {
             get_message_descriptor(&self.protobuf.desc_file, &self.protobuf.message_type)?;
         Ok(ProtobufSerializer {
             message_descriptor,
-            options: Options::default(),
+            options: Options {
+                use_json_names: self.protobuf.use_json_names,
+            },
         })
     }
 
@@ -62,6 +64,16 @@ pub struct ProtobufSerializerOptions {
     /// The name of the message type to use for serializing.
     #[configurable(metadata(docs::examples = "package.Message"))]
     pub message_type: String,
+
+    /// Use JSON field names (camelCase) instead of protobuf field names (snake_case).
+    ///
+    /// When enabled, the serializer looks for fields using their JSON names as defined
+    /// in the `.proto` file (for example `jobDescription` instead of `job_description`).
+    ///
+    /// This is useful when working with data that has already been converted from JSON or
+    /// when interfacing with systems that use JSON naming conventions.
+    #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
+    pub use_json_names: bool,
 }
 
 /// Serializer that converts an `Event` to bytes using the Protobuf format.

lib/codecs/tests/protobuf.rs

@@ -26,11 +26,13 @@ fn read_protobuf_bin_message(path: &Path) -> Bytes {
 fn build_serializer_pair(
     desc_file: PathBuf,
     message_type: String,
+    use_json_names: bool,
 ) -> (ProtobufSerializer, ProtobufDeserializer) {
     let serializer = ProtobufSerializerConfig {
         protobuf: ProtobufSerializerOptions {
             desc_file: desc_file.clone(),
             message_type: message_type.clone(),
+            use_json_names,
         },
     }
     .build()
@@ -39,6 +41,7 @@ fn build_serializer_pair(
         protobuf: ProtobufDeserializerOptions {
             desc_file,
             message_type,
+            use_json_names,
         },
     }
     .build()
@@ -52,7 +55,7 @@ fn roundtrip_coding() {
         read_protobuf_bin_message(&test_data_dir().join("pbs/person_someone.pb"));
     let desc_file = test_data_dir().join("protos/test_protobuf.desc");
     let message_type: String = "test_protobuf.Person".into();
-    let (mut serializer, deserializer) = build_serializer_pair(desc_file, message_type);
+    let (mut serializer, deserializer) = build_serializer_pair(desc_file, message_type, false);
 
     let events_original = deserializer
         .parse(protobuf_message, LogNamespace::Vector)
@@ -68,3 +71,79 @@ fn roundtrip_coding() {
         .unwrap();
     assert_eq!(events_original, events_encoded);
 }
+
+#[test]
+fn roundtrip_coding_with_json_names() {
+    let protobuf_message =
+        read_protobuf_bin_message(&test_data_dir().join("pbs/person_someone3.pb"));
+    let desc_file = test_data_dir().join("protos/test_protobuf3.desc");
+    let message_type: String = "test_protobuf3.Person".into();
+
+    // Test with use_json_names=false (default behavior - snake_case field names)
+    let (mut serializer_snake_case, deserializer_snake_case) =
+        build_serializer_pair(desc_file.clone(), message_type.clone(), false);
+
+    let events_snake_case = deserializer_snake_case
+        .parse(protobuf_message.clone(), LogNamespace::Vector)
+        .unwrap();
+    assert_eq!(1, events_snake_case.len());
+
+    // Verify that protobuf field names are being used (snake_case)
+    let event = events_snake_case[0].as_log();
+    assert!(
+        event.contains("job_description"),
+        "Event should contain 'job_description' (protobuf field name) when use_json_names is disabled"
+    );
+    assert_eq!(
+        event.get("job_description").unwrap().to_string_lossy(),
+        "Software Engineer"
+    );
+    assert!(
+        !event.contains("jobDescription"),
+        "Event should not contain 'jobDescription' (JSON name) when use_json_names is disabled"
+    );
+
+    // Test roundtrip with snake_case
+    let mut new_message = BytesMut::new();
+    serializer_snake_case
+        .encode(events_snake_case[0].clone(), &mut new_message)
+        .unwrap();
+    let events_encoded = deserializer_snake_case
+        .parse(new_message.into(), LogNamespace::Vector)
+        .unwrap();
+    assert_eq!(events_snake_case, events_encoded);
+
+    // Test with use_json_names=true (camelCase field names)
+    let (mut serializer_camel_case, deserializer_camel_case) =
+        build_serializer_pair(desc_file, message_type, true);
+
+    let events_camel_case = deserializer_camel_case
+        .parse(protobuf_message, LogNamespace::Vector)
+        .unwrap();
+    assert_eq!(1, events_camel_case.len());
+
+    // Verify that JSON names are being used (camelCase)
+    let event = events_camel_case[0].as_log();
+    assert!(
+        event.contains("jobDescription"),
+        "Event should contain 'jobDescription' (JSON name) when use_json_names is enabled"
+    );
+    assert_eq!(
+        event.get("jobDescription").unwrap().to_string_lossy(),
+        "Software Engineer"
+    );
+    assert!(
+        !event.contains("job_description"),
+        "Event should not contain 'job_description' (protobuf name) when use_json_names is enabled"
+    );
+
+    // Test roundtrip with camelCase
+    let mut new_message = BytesMut::new();
+    serializer_camel_case
+        .encode(events_camel_case[0].clone(), &mut new_message)
+        .unwrap();
+    let events_encoded = deserializer_camel_case
+        .parse(new_message.into(), LogNamespace::Vector)
+        .unwrap();
+    assert_eq!(events_camel_case, events_encoded);
+}

src/components/validation/resources/mod.rs

@@ -157,6 +157,7 @@ fn deserializer_config_to_serializer(config: &DeserializerConfig) -> encoding::S
                 protobuf: vector_lib::codecs::encoding::ProtobufSerializerOptions {
                     desc_file: config.protobuf.desc_file.clone(),
                     message_type: config.protobuf.message_type.clone(),
+                    use_json_names: config.protobuf.use_json_names,
                 },
             })
         }
@@ -229,6 +230,7 @@ fn serializer_config_to_deserializer(
                 protobuf: vector_lib::codecs::decoding::ProtobufDeserializerOptions {
                     desc_file: config.protobuf.desc_file.clone(),
                     message_type: config.protobuf.message_type.clone(),
+                    use_json_names: config.protobuf.use_json_names,
                 },
             })
         }

src/sinks/util/encoding.rs

@@ -406,6 +406,7 @@ mod tests {
             protobuf: ProtobufSerializerOptions {
                 desc_file: test_data_dir().join("test_proto.desc"),
                 message_type: "test_proto.User".to_string(),
+                use_json_names: false,
             },
         };
 
@@ -460,6 +461,7 @@ mod tests {
             protobuf: ProtobufSerializerOptions {
                 desc_file: test_data_dir().join("test_proto.desc"),
                 message_type: "test_proto.User".to_string(),
+                use_json_names: false,
             },
         };
 

website/content/en/highlights/2025-09-23-otlp-support.md

@@ -44,19 +44,9 @@ sinks:
       uri: http://otel-collector-sink:5318/v1/logs
       method: post
       encoding:
-        codec: json
-      framing:
-        method: newline_delimited
-      batch:
-        max_events: 1
-      request:
-        headers:
-          content-type: application/json
+        codec: otlp
 ```
 
-**Note:** This setup is affected by a [known issue](https://github.com/vectordotdev/vector/issues/22054).
-We plan to improve batching for this sink in future Vector versions.
-
 ## Example Configuration 2
 
 Here is another pipeline configuration that can achieve the same as the above:
@@ -75,6 +65,7 @@ otel_sink:
       protobuf:
         desc_file: path/to/opentelemetry-proto.desc
         message_type: opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest
+        use_json_names: true
     framing:
       method: 'bytes'
     request:
@@ -91,5 +82,3 @@ The `desc` file was generated with the following command:
     --descriptor_set_out=opentelemetry-proto.desc \\
           $(find /path/to/vector/lib/opentelemetry-proto/src/proto/opentelemetry-proto -name '*.proto')
 ```
-
-**Note:** In the future, we can simplify the `opentelemetry` sink UX further, eliminating the need to compile proto files.

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

@@ -397,6 +397,19 @@ generated: components: sinks: amqp: configuration: {
 						required:    true
 						type: string: examples: ["package.Message"]
 					}
+					use_json_names: {
+						description: """
+																Use JSON field names (camelCase) instead of protobuf field names (snake_case).
+
+																When enabled, the serializer looks for fields using their JSON names as defined
+																in the `.proto` file (for example `jobDescription` instead of `job_description`).
+
+																This is useful when working with data that has already been converted from JSON or
+																when interfacing with systems that use JSON naming conventions.
+																"""
+						required: false
+						type: bool: default: false
+					}
 				}
 			}
 			timestamp_format: {