DP-2782: Backport evolution bugfix to Connector 1 (DP-2767) (#61)

celonis · Feb 13, 2024 · cb95933 · cb95933
1 parent 101b6c0
commit cb95933
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 0 deletions.
diff --git a/connector/src/main/java/com/celonis/kafka/connect/schema/StructSchemaEvolution.java b/connector/src/main/java/com/celonis/kafka/connect/schema/StructSchemaEvolution.java
@@ -75,6 +75,8 @@ private Schema mergeStructs(Schema currentSchema, Schema recordSchema)
       throws SchemaEvolutionException {
     SchemaBuilder result = SchemaUtils.withMetadata(SchemaBuilder.struct(), currentSchema);
 
+    if (currentSchema.isOptional() || recordSchema.isOptional()) result.optional();
+
     // First currentSchemaFields
     currentSchema.fields().stream()
         .forEach(

diff --git a/connector/src/test/scala/com/celonis/kafka/connect/schema/StructSchemaEvolutionTest.scala b/connector/src/test/scala/com/celonis/kafka/connect/schema/StructSchemaEvolutionTest.scala
@@ -183,6 +183,28 @@ class StructSchemaEvolutionTest extends AnyFunSuite with Matchers with Inside {
 
     schemaEv.evolve(currentSchema, recordSchema) shouldEqual expectedResult
   }
+
+  test("testSchemaEvolutionPreservesFieldOptionality") {
+    val currentSchema =
+      SchemaBuilder.struct.field("a_struct",
+                                 SchemaBuilder.struct.field("optional_string", Schema.OPTIONAL_STRING_SCHEMA).build,
+      ).field("an_empty_struct", SchemaBuilder.struct.optional.schema).build
+    val recordSchema = SchemaBuilder.struct.field(
+      "a_struct",
+      SchemaBuilder.struct.optional.field("optional_string", Schema.OPTIONAL_STRING_SCHEMA).field("an_int",
+                                                                                                  Schema.INT64_SCHEMA,
+      ).field("a_string_2", Schema.STRING_SCHEMA).build,
+    ).field("an_empty_struct", SchemaBuilder.struct.schema).build
+    val expectedResult = SchemaBuilder.struct.field(
+      "a_struct",
+      SchemaBuilder.struct.optional.field("optional_string", Schema.OPTIONAL_STRING_SCHEMA).field("an_int",
+                                                                                                  Schema.INT64_SCHEMA,
+      ).field("a_string_2", Schema.STRING_SCHEMA).build,
+    ).field("an_empty_struct", SchemaBuilder.struct.optional.schema).build
+    val result = schemaEv.evolve(currentSchema, recordSchema)
+    result shouldEqual expectedResult
+  }
+
   test("SchemaEvolutionWithAdditionalStructRecordField") {
     val currentSchema =
       SchemaBuilder.struct()