[Fix][Formats][Protobuf] Ensure case-sensitive field names are preserved in Protobuf serialization and deserialization

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java

@@ -1623,6 +1623,177 @@ public void testKafkaProtobufForTransformToAssert(TestContainer container)
         }
     }
 
+    @TestTemplate
+    public void testProtobufCaseSensitiveFieldNames(TestContainer container)
+            throws IOException, InterruptedException, URISyntaxException {
+        Container.ExecResult execResult =
+                container.executeJob("/protobuf/fake_to_kafka_protobuf_case_sensitive.conf");
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
+
+        String path = getTestConfigFile("/protobuf/fake_to_kafka_protobuf_case_sensitive.conf");
+        Config config = ConfigFactory.parseFile(new File(path));
+        Config sinkConfig = config.getConfigList("sink").get(0);
+
+        Map<String, String> schemaProperties = new HashMap<>();
+        schemaProperties.put(
+                "protobuf_message_name", sinkConfig.getString("protobuf_message_name"));
+        schemaProperties.put("protobuf_schema", sinkConfig.getString("protobuf_schema"));
+
+        SeaTunnelRowType nestedType =
+                new SeaTunnelRowType(
+                        new String[] {"NestedField", "AnotherField"},
+                        new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE, BasicType.INT_TYPE});
+
+        SeaTunnelRowType seaTunnelRowType =
+                new SeaTunnelRowType(
+                        new String[] {
+                            "MyIntField",
+                            "CamelCaseString",
+                            "snake_case_field",
+                            "NestedObject",
+                            "MyMapField"
+                        },
+                        new SeaTunnelDataType<?>[] {
+                            BasicType.INT_TYPE,
+                            BasicType.STRING_TYPE,
+                            BasicType.STRING_TYPE,
+                            nestedType,
+                            new MapType<>(BasicType.STRING_TYPE, BasicType.INT_TYPE)
+                        });
+
+        TableSchema schema =
+                TableSchema.builder()
+                        .columns(
+                                Arrays.asList(
+                                        IntStream.range(0, seaTunnelRowType.getTotalFields())
+                                                .mapToObj(
+                                                        i ->
+                                                                PhysicalColumn.of(
+                                                                        seaTunnelRowType
+                                                                                .getFieldName(i),
+                                                                        seaTunnelRowType
+                                                                                .getFieldType(i),
+                                                                        0,
+                                                                        true,
+                                                                        null,
+                                                                        null))
+                                                .toArray(PhysicalColumn[]::new)))
+                        .build();
+
+        CatalogTable catalogTable =
+                CatalogTable.of(
+                        TableIdentifier.of("", "", "", "test"),
+                        schema,
+                        schemaProperties,
+                        Collections.emptyList(),
+                        "It is converted from RowType and only has column information.");
+
+        ProtobufDeserializationSchema deserializationSchema =
+                new ProtobufDeserializationSchema(catalogTable);
+
+        List<SeaTunnelRow> kafkaSTRow =
+                getKafkaSTRow(
+                        "test_protobuf_case_sensitive_topic",
+                        value -> {
+                            try {
+                                return deserializationSchema.deserialize(value);
+                            } catch (IOException e) {
+                                throw new RuntimeException("Error deserializing Kafka message", e);
+                            }
+                        });
+
+        Assertions.assertEquals(16, kafkaSTRow.size());
+
+        kafkaSTRow.forEach(
+                row -> {
+                    Assertions.assertAll(
+                            "Verify case-sensitive field values",
+                            () -> Assertions.assertNotNull(row.getField(0)), // MyIntField
+                            () -> Assertions.assertNotNull(row.getField(1)), // CamelCaseString
+                            () -> Assertions.assertNotNull(row.getField(2)), // snake_case_field
+                            () -> {
+                                SeaTunnelRow nestedRow = (SeaTunnelRow) row.getField(3);
+                                if (nestedRow != null) {
+                                    Assertions.assertNotNull(nestedRow.getField(0)); // NestedField
+                                    Assertions.assertNotNull(nestedRow.getField(1)); // AnotherField
+                                }
+                            },
+                            () -> {
+                                @SuppressWarnings("unchecked")
+                                Map<String, Integer> mapField =
+                                        (Map<String, Integer>) row.getField(4);
+                                if (mapField != null) {
+                                    Assertions.assertNotNull(mapField);
+                                }
+                            });
+                });
+    }
+
+    @TestTemplate
+    public void testProtobufCaseSensitiveToAssert(TestContainer container)
+            throws IOException, InterruptedException, URISyntaxException {
+
+        String confFile = "/protobuf/kafka_protobuf_case_sensitive_to_assert.conf";
+        String path = getTestConfigFile(confFile);
+        Config config = ConfigFactory.parseFile(new File(path));
+        Config sourceConfig = config.getConfigList("source").get(0);
+        ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(sourceConfig);
+
+        SeaTunnelRowType nestedType =
+                new SeaTunnelRowType(
+                        new String[] {"NestedField", "AnotherField"},
+                        new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE, BasicType.INT_TYPE});
+
+        SeaTunnelRowType seaTunnelRowType =
+                new SeaTunnelRowType(
+                        new String[] {
+                            "MyIntField",
+                            "CamelCaseString",
+                            "snake_case_field",
+                            "NestedObject",
+                            "MyMapField"
+                        },
+                        new SeaTunnelDataType<?>[] {
+                            BasicType.INT_TYPE,
+                            BasicType.STRING_TYPE,
+                            BasicType.STRING_TYPE,
+                            nestedType,
+                            new MapType<>(BasicType.STRING_TYPE, BasicType.INT_TYPE)
+                        });
+
+        DefaultSeaTunnelRowSerializer serializer =
+                getDefaultSeaTunnelRowSerializer(
+                        "test_protobuf_case_sensitive_topic", seaTunnelRowType, readonlyConfig);
+
+        SeaTunnelRow nestedRow = new SeaTunnelRow(2);
+        nestedRow.setField(0, "nested_value");
+        nestedRow.setField(1, 999);
+
+        Map<String, Integer> mapData = new HashMap<>();
+        mapData.put("key1", 100);
+        mapData.put("key2", 200);
+
+        for (int i = 0; i < 16; i++) {
+            SeaTunnelRow row = new SeaTunnelRow(5);
+            row.setField(0, i);
+            row.setField(1, "test_string_" + i);
+            row.setField(2, "snake_value_" + i);
+            row.setField(3, nestedRow);
+            row.setField(4, mapData);
+
+            ProducerRecord<byte[], byte[]> producerRecord = serializer.serializeRow(row);
+            try {
+                producer.send(producerRecord).get();
+            } catch (InterruptedException | ExecutionException e) {
+                throw new RuntimeException("Error sending Kafka message", e);
+            }
+        }
+        producer.flush();
+
+        Container.ExecResult execResult = container.executeJob(confFile);
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
+    }
+
     public static String getTestConfigFile(String configFile)
             throws FileNotFoundException, URISyntaxException {
         URL resource = KafkaIT.class.getResource(configFile);

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/protobuf/fake_to_kafka_protobuf_case_sensitive.conf

@@ -0,0 +1,86 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+    parallelism = 1
+    job.mode = "BATCH"
+
+    # spark config
+    spark.executor.instances = 1
+    spark.executor.cores = 1
+    spark.executor.memory = "1g"
+    spark.master = local
+
+}
+source {
+   FakeSource {
+      parallelism = 1
+      plugin_output = "fake"
+      row.num = 16
+      schema = {
+        fields {
+                  MyIntField = int
+                  CamelCaseString = string
+                  snake_case_field = string
+
+                  NestedObject {
+                      NestedField = string
+                      AnotherField = int
+                  }
+                  MyMapField = "map<string,int>"
+        }
+      }
+    }
+}
+
+sink {
+  kafka {
+      topic = "test_protobuf_case_sensitive_topic"
+      bootstrap.servers = "kafkaCluster:9092"
+      format = protobuf
+      kafka.request.timeout.ms = 60000
+      kafka.config = {
+        acks = "all"
+        request.timeout.ms = 60000
+        buffer.memory = 33554432
+      }
+      protobuf_message_name = TestCaseSensitive
+      protobuf_schema = """
+              syntax = "proto3";
+
+              package org.apache.seatunnel.format.protobuf;
+
+              option java_outer_classname = "ProtobufCaseSensitiveE2E";
+
+              message TestCaseSensitive {
+                int32 MyIntField = 1;
+                string CamelCaseString = 2;
+                string snake_case_field = 3;
+
+                message NestedObject {
+                  string NestedField = 1;
+                  int32 AnotherField = 2;
+                }
+
+                NestedObject nestedObject = 4;
+
+                map<string, int32> MyMapField = 5;
+              }
+              """
+  }
+}
+

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/protobuf/kafka_protobuf_case_sensitive_to_assert.conf

@@ -0,0 +1,106 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+    parallelism = 1
+    job.mode = "BATCH"
+
+    # spark config
+    spark.executor.instances = 1
+    spark.executor.cores = 1
+    spark.executor.memory = "1g"
+    spark.master = local
+}
+
+source {
+    Kafka {
+        topic = "test_protobuf_case_sensitive_topic"
+        format = protobuf
+        protobuf_message_name = TestCaseSensitive
+        protobuf_schema = """
+            syntax = "proto3";
+
+            package org.apache.seatunnel.format.protobuf;
+
+            option java_outer_classname = "ProtobufCaseSensitiveE2E";
+
+            message TestCaseSensitive {
+                int32 MyIntField = 1;
+                string CamelCaseString = 2;
+                string snake_case_field = 3;
+
+                message NestedObject {
+                    string NestedField = 1;
+                    int32 AnotherField = 2;
+                }
+
+                NestedObject nestedObject = 4;
+
+                map<string, int32> MyMapField = 5;
+            }
+        """
+        schema = {
+            fields {
+                MyIntField = int
+                CamelCaseString = string
+                snake_case_field = string
+
+                NestedObject {
+                    NestedField = string
+                    AnotherField = int
+                }
+                MyMapField = "map<string,int>"
+            }
+        }
+        bootstrap.servers = "kafkaCluster:9092"
+        start_mode = "earliest"
+        plugin_output = "kafka_table"
+    }
+}
+
+sink {
+    Assert {
+        plugin_input = "kafka_table"
+        rules {
+            row_rules = [
+                {
+                    rule_type = MAX_ROW
+                    rule_value = 16
+                },
+                {
+                    rule_type = MIN_ROW
+                    rule_value = 16
+                }
+            ],
+            field_rules = [
+                {
+                    field_name = MyIntField
+                    field_type = int
+                },
+                {
+                    field_name = CamelCaseString
+                    field_type = string
+                },
+                {
+                    field_name = snake_case_field
+                    field_type = string
+                }
+            ]
+        }
+    }
+}
+

seatunnel-formats/seatunnel-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/RowToProtobufConverter.java

@@ -56,8 +56,7 @@ public byte[] convertRowToGenericRecord(SeaTunnelRow element) {
                 if (resolvedValue instanceof byte[]) {
                     resolvedValue = ByteString.copyFrom((byte[]) resolvedValue);
                 }
-                builder.setField(
-                        descriptor.findFieldByName(fieldName.toLowerCase()), resolvedValue);
+                builder.setField(descriptor.findFieldByName(fieldName), resolvedValue);
             }
         }