From 313f559ad2ea8667f008c8e1e81439244d3634c9 Mon Sep 17 00:00:00 2001
From: Chris Baumann <christopher.baumann@konghq.com>
Date: Wed, 23 Oct 2024 17:27:31 +0200
Subject: [PATCH] test(AvroConverter): add test case for single field
 serialization

Adds test case for single field serialization that shows that the order of fields in the input schema affects the output.
---
 .../connect/avro/AvroConverterTest.java       | 49 +++++++++++++++++++
 1 file changed, 49 insertions(+)

diff --git a/avro-converter/src/test/java/io/confluent/connect/avro/AvroConverterTest.java b/avro-converter/src/test/java/io/confluent/connect/avro/AvroConverterTest.java
index 12e0f7798d6..e606b31b10a 100644
--- a/avro-converter/src/test/java/io/confluent/connect/avro/AvroConverterTest.java
+++ b/avro-converter/src/test/java/io/confluent/connect/avro/AvroConverterTest.java
@@ -102,6 +102,55 @@ public void testPrimitive() {
     assertEquals(expected, schemaAndValue);
   }
 
+  @Test
+  public void testSingleFieldSerialization() throws RestClientException, IOException {
+    SchemaRegistryClient schemaRegistry = new MockSchemaRegistryClient();
+    AvroConverter avroConverter = new AvroConverter(schemaRegistry);
+    Map<String, ?> converterConfig = ImmutableMap.of(
+            AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "localhost",
+            AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS, false,
+            AbstractKafkaSchemaSerDeConfig.LATEST_COMPATIBILITY_STRICT, false,
+            AbstractKafkaSchemaSerDeConfig.NORMALIZE_SCHEMAS, true,
+            AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION, true,
+            AbstractKafkaSchemaSerDeConfig.VALUE_SUBJECT_NAME_STRATEGY, TopicNameStrategy.class.getName());
+    avroConverter.configure(converterConfig, false);
+
+    org.apache.avro.Schema registredSchema = org.apache.avro.SchemaBuilder
+            .record("MySchema")
+            .fields()
+            .requiredString("id")
+            .endRecord();
+
+    schemaRegistry.register("topic-value", new AvroSchema(registredSchema));
+
+    Schema inputSchema1 = SchemaBuilder.struct()
+            .field("foo", org.apache.kafka.connect.data.Schema.STRING_SCHEMA)
+            .field("id", org.apache.kafka.connect.data.Schema.STRING_SCHEMA)
+            .build();
+
+    Struct inputValue1 = new Struct(inputSchema1)
+            .put("foo", "123")
+            .put("id", "456");
+
+    Schema inputSchema2 = SchemaBuilder.struct()
+            .field("id", org.apache.kafka.connect.data.Schema.STRING_SCHEMA)
+            .field("foo", org.apache.kafka.connect.data.Schema.STRING_SCHEMA)
+            .build();
+
+    Struct inputValue2 = new Struct(inputSchema2)
+            .put("id", "456")
+            .put("foo", "123");
+
+    final byte[] bytes1 = avroConverter.fromConnectData("topic", inputSchema1, inputValue1);
+    final SchemaAndValue schemaAndValue1 = avroConverter.toConnectData("topic", bytes1);
+
+    final byte[] bytes2 = avroConverter.fromConnectData("topic", inputSchema2, inputValue2);
+    final SchemaAndValue schemaAndValue2 = avroConverter.toConnectData("topic", bytes2);
+
+
+    assertEquals(schemaAndValue1.value(), schemaAndValue2.value());
+  }
+
   @Test
   public void testComplex() {
     SchemaBuilder builder = SchemaBuilder.struct()