From e51834015518037de2aa694f927cf9d35af45f40 Mon Sep 17 00:00:00 2001
From: Ismail Simsek <gel.yine.gel@hotmail.com>
Date: Wed, 5 Jan 2022 11:40:58 +0100
Subject: [PATCH] improve testing helper classes

---
 .run/IcebergChangeConsumerTest.run.xml        |   9 -
 ...ChangeConsumerUpsertDeleteDeletesTest.java | 130 ++++---------
 .../IcebergChangeConsumerUpsertTest.java      | 154 +++++-----------
 .../IcebergChangeEventBuilderTest.java        |  75 ++++++++
 .../IcebergChangeEventBuilder.java            | 171 ++++++++++++++++++
 .../testresources/TestChangeEvent.java        |  84 +++++++++
 6 files changed, 412 insertions(+), 211 deletions(-)
 create mode 100644 debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/tableoperator/IcebergChangeEventBuilderTest.java
 create mode 100644 debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/IcebergChangeEventBuilder.java

diff --git a/.run/IcebergChangeConsumerTest.run.xml b/.run/IcebergChangeConsumerTest.run.xml
index 8ccfa875..12eeb322 100644
--- a/.run/IcebergChangeConsumerTest.run.xml
+++ b/.run/IcebergChangeConsumerTest.run.xml
@@ -1,11 +1,3 @@
-<!--
-  ~ /*
-  ~  * Copyright memiiso Authors.
-  ~  *
-  ~  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
-  ~  */
-  -->
-
 <component name="ProjectRunConfigurationManager">
     <configuration default="false" name="IcebergChangeConsumerTest" type="JUnit" factoryName="JUnit">
         <module name="debezium-server-iceberg-sink"/>
@@ -19,7 +11,6 @@
         <option name="MAIN_CLASS_NAME" value="io.debezium.server.iceberg.IcebergChangeConsumerTest"/>
         <option name="METHOD_NAME" value=""/>
         <option name="TEST_OBJECT" value="class"/>
-        <option name="VM_PARAMETERS" value="-ea -Ddebezium.sink.type=iceberg"/>
         <option name="WORKING_DIRECTORY" value="$MODULE_DIR$"/>
         <method v="2">
             <option name="Make" enabled="true"/>
diff --git a/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerUpsertDeleteDeletesTest.java b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerUpsertDeleteDeletesTest.java
index 3b1f0671..d09ae1d7 100644
--- a/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerUpsertDeleteDeletesTest.java
+++ b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerUpsertDeleteDeletesTest.java
@@ -8,13 +8,11 @@
 
 package io.debezium.server.iceberg;
 
-import io.debezium.engine.ChangeEvent;
 import io.debezium.server.iceberg.testresources.*;
 import io.quarkus.test.common.QuarkusTestResource;
 import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.junit.TestProfile;
 
-import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import javax.inject.Inject;
@@ -41,10 +39,11 @@ public class IcebergChangeConsumerUpsertDeleteDeletesTest extends BaseSparkTest
   @Test
   public void testSimpleUpsert() throws Exception {
 
-    List<ChangeEvent<Object, Object>> records = new ArrayList<>();
-    records.add(getCustomerRecord(1, "c"));
-    records.add(getCustomerRecord(2, "c"));
-    records.add(getCustomerRecord(3, "c"));
+    String dest = "inventory.customers_upsert";
+    List<io.debezium.engine.ChangeEvent<Object, Object>> records = new ArrayList<>();
+    records.add(TestChangeEvent.of(dest, 1, "c"));
+    records.add(TestChangeEvent.of(dest, 2, "c"));
+    records.add(TestChangeEvent.of(dest, 3, "c"));
     consumer.handleBatch(records, TestUtil.getCommitter());
 
     Dataset<Row> ds = getTableData("testc.inventory.customers_upsert");
@@ -53,10 +52,10 @@ public void testSimpleUpsert() throws Exception {
 
     // 3 records should be updated 4th one should be inserted
     records.clear();
-    records.add(getCustomerRecord(1, "r"));
-    records.add(getCustomerRecord(2, "d"));
-    records.add(getCustomerRecord(3, "u", "UpdatednameV1"));
-    records.add(getCustomerRecord(4, "c"));
+    records.add(TestChangeEvent.of(dest, 1, "r"));
+    records.add(TestChangeEvent.of(dest, 2, "d"));
+    records.add(TestChangeEvent.of(dest, 3, "u", "UpdatednameV1"));
+    records.add(TestChangeEvent.of(dest, 4, "c"));
     consumer.handleBatch(records, TestUtil.getCommitter());
 
     ds = getTableData("testc.inventory.customers_upsert");
@@ -70,17 +69,17 @@ public void testSimpleUpsert() throws Exception {
 
     records.clear();
     // incase of duplicate records it should only keep the latest by epoch ts
-    records.add(getCustomerRecord(3, "r", "UpdatednameV2", 1L));
-    records.add(getCustomerRecord(3, "u", "UpdatednameV3", 2L));
-    records.add(getCustomerRecord(3, "u", "UpdatednameV4", 3L));
-    records.add(getCustomerRecord(4, "u", "Updatedname-4-V1", 4L));
-    records.add(getCustomerRecord(4, "u", "Updatedname-4-V2", 5L));
-    records.add(getCustomerRecord(4, "d", "Updatedname-4-V3", 6L));
-    records.add(getCustomerRecord(5, "d", 7L));
-    records.add(getCustomerRecord(6, "r", 8L));
-    records.add(getCustomerRecord(6, "r", 9L));
-    records.add(getCustomerRecord(6, "u", 10L));
-    records.add(getCustomerRecord(6, "u", "Updatedname-6-V1", 11L));
+    records.add(TestChangeEvent.of(dest, 3, "r", "UpdatednameV2", 1L));
+    records.add(TestChangeEvent.of(dest, 3, "u", "UpdatednameV3", 2L));
+    records.add(TestChangeEvent.of(dest, 3, "u", "UpdatednameV4", 3L));
+    records.add(TestChangeEvent.of(dest, 4, "u", "Updatedname-4-V1", 4L));
+    records.add(TestChangeEvent.of(dest, 4, "u", "Updatedname-4-V2", 5L));
+    records.add(TestChangeEvent.of(dest, 4, "d", "Updatedname-4-V3", 6L));
+    records.add(TestChangeEvent.of(dest, 5, "d", 7L));
+    records.add(TestChangeEvent.of(dest, 6, "r", 8L));
+    records.add(TestChangeEvent.of(dest, 6, "r", 9L));
+    records.add(TestChangeEvent.of(dest, 6, "u", 10L));
+    records.add(TestChangeEvent.of(dest, 6, "u", "Updatedname-6-V1", 11L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     ds = getTableData("testc.inventory.customers_upsert");
     ds.show();
@@ -93,10 +92,10 @@ public void testSimpleUpsert() throws Exception {
     // in case of duplicate records including epoch ts, its should keep latest one based on operation priority
     // ("c", 1, "r", 2, "u", 3, "d", 4);
     records.clear();
-    records.add(getCustomerRecord(3, "d", "UpdatednameV5", 1L));
-    records.add(getCustomerRecord(3, "u", "UpdatednameV6", 1L));
-    records.add(getCustomerRecord(6, "c", "Updatedname-6-V2", 1L));
-    records.add(getCustomerRecord(6, "r", "Updatedname-6-V3", 1L));
+    records.add(TestChangeEvent.of(dest, 3, "d", "UpdatednameV5", 1L));
+    records.add(TestChangeEvent.of(dest, 3, "u", "UpdatednameV6", 1L));
+    records.add(TestChangeEvent.of(dest, 6, "c", "Updatedname-6-V2", 1L));
+    records.add(TestChangeEvent.of(dest, 6, "r", "Updatedname-6-V3", 1L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     ds = getTableData("testc.inventory.customers_upsert");
     ds.show();
@@ -105,10 +104,10 @@ public void testSimpleUpsert() throws Exception {
 
     // if its not standard insert followed by update! should keep latest one
     records.clear();
-    records.add(getCustomerRecord(7, "u", 1L));
-    records.add(getCustomerRecord(7, "d", 2L));
-    records.add(getCustomerRecord(7, "r", 3L));
-    records.add(getCustomerRecord(7, "u", "Updatedname-7-V1", 4L));
+    records.add(TestChangeEvent.of(dest, 7, "u", 1L));
+    records.add(TestChangeEvent.of(dest, 7, "d", 2L));
+    records.add(TestChangeEvent.of(dest, 7, "r", 3L));
+    records.add(TestChangeEvent.of(dest, 7, "u", "Updatedname-7-V1", 4L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     ds = getTableData("testc.inventory.customers_upsert");
     ds.show();
@@ -118,12 +117,13 @@ public void testSimpleUpsert() throws Exception {
 
   @Test
   public void testSimpleUpsertCompositeKey() throws Exception {
+    String dest = "inventory.customers_upsert_compositekey";
     // test simple inserts
-    List<ChangeEvent<Object, Object>> records = new ArrayList<>();
-    records.add(getCustomerRecordCompositeKey(1, "c", "user1", 1L));
-    records.add(getCustomerRecordCompositeKey(1, "c", "user2", 1L));
-    records.add(getCustomerRecordCompositeKey(1, "u", "user1", 2L));
-    records.add(getCustomerRecordCompositeKey(1, "r", "user1", 3L));
+    List<io.debezium.engine.ChangeEvent<Object, Object>> records = new ArrayList<>();
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "c", "user1", 1L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "c", "user2", 1L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "u", "user1", 2L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "r", "user1", 3L));
     consumer.handleBatch(records, TestUtil.getCommitter());
 
     Dataset<Row> ds = getTableData("testc.inventory.customers_upsert_compositekey");
@@ -132,67 +132,15 @@ public void testSimpleUpsertCompositeKey() throws Exception {
     Assertions.assertEquals(ds.where("id = 1").count(), 2);
 
     records.clear();
-    records.add(getCustomerRecordCompositeKey(1, "u", "user1", 2L));
-    records.add(getCustomerRecordCompositeKey(1, "r", "user1", 3L));
-    records.add(getCustomerRecordCompositeKey(1, "d", "user1", 3L));
-    records.add(getCustomerRecordCompositeKey(1, "d", "user2", 1L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "u", "user1", 2L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "r", "user1", 3L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "d", "user1", 3L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "d", "user2", 1L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     ds = getTableData("testc.inventory.customers_upsert_compositekey");
     ds.show();
     Assertions.assertEquals(ds.count(), 0);
     Assertions.assertEquals(ds.where("first_name= 'user2'").count(), 0);
   }
-
-  private TestChangeEvent<Object, Object> getCustomerRecord(Integer id, String operation, String name, Long epoch) {
-    String key = "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false," + "\"field\":\"id\"}]," +
-                 "\"optional\":false,\"name\":\"testc.inventory.customers.Key\"}," +
-                 "\"payload\":{\"id\":" + id + "}}";
-    String val = "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"field\":\"id\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"first_name\"},{\"type\":\"string\",\"optional\":false,\"field\":\"last_name\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"email\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__op\"}," +
-                 "{\"type\":\"string\",\"optional\":true,\"field\":\"__table\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"__lsn\"}," +
-                 "{\"type\":\"int64\",\"optional\":true,\"field\":\"__source_ts_ms\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__deleted\"}]," +
-                 "\"optional\":false,\"name\":\"testc.inventory.customers.Value\"}," +
-                 "\"payload\":{\"id\":" + id + ",\"first_name\":\"" + name + "\",\"last_name\":\"Walker\",\"email\":\"ed@walker" +
-                 ".com\"," +
-                 "\"__op\":\"" + operation + "\",\"__table\":\"customers\",\"__lsn\":33832960,\"__source_ts_ms\":" + epoch + "," +
-                 "\"__deleted\":\"" + operation.equals("d") + "\"}} ";
-    return new TestChangeEvent<>(key, val, "testc.inventory.customers_upsert");
-  }
-
-  private TestChangeEvent<Object, Object> getCustomerRecordCompositeKey(Integer id, String operation, String name,
-                                                                        Long epoch) {
-    String key = "{\"schema\":{\"type\":\"struct\",\"fields\":[" +
-                 "{\"type\":\"int32\",\"optional\":false," + "\"field\":\"id\"}," +
-                 "{\"type\":\"string\",\"optional\":false," + "\"field\":\"first_name\"}" +
-                 "]," +
-                 "\"optional\":false,\"name\":\"testc.inventory.customers.Key\"}," +
-                 "\"payload\":{\"id\":" + id + ",\"first_name\":\"" + name + "\"}}";
-
-    String val = "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"field\":\"id\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"first_name\"},{\"type\":\"string\",\"optional\":false,\"field\":\"last_name\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"email\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__op\"}," +
-                 "{\"type\":\"string\",\"optional\":true,\"field\":\"__table\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"__lsn\"}," +
-                 "{\"type\":\"int64\",\"optional\":true,\"field\":\"__source_ts_ms\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__deleted\"}]," +
-                 "\"optional\":false,\"name\":\"testc.inventory.customers.Value\"}," +
-                 "\"payload\":{\"id\":" + id + ",\"first_name\":\"" + name + "\",\"last_name\":\"Walker\",\"email\":\"ed@walker" +
-                 ".com\"," +
-                 "\"__op\":\"" + operation + "\",\"__table\":\"customers\",\"__lsn\":33832960,\"__source_ts_ms\":" + epoch + "," +
-                 "\"__deleted\":\"" + operation.equals("d") + "\"}} ";
-    return new TestChangeEvent<>(key, val, "testc.inventory.customers_upsert_compositekey");
-  }
-
-  private TestChangeEvent<Object, Object> getCustomerRecord(Integer id, String operation) {
-    return this.getCustomerRecord(id, operation, TestUtil.randomString(12), Instant.now().toEpochMilli());
-  }
-
-
-  private TestChangeEvent<Object, Object> getCustomerRecord(Integer id, String operation, String name) {
-    return this.getCustomerRecord(id, operation, name, Instant.now().toEpochMilli());
-  }
-
-  private TestChangeEvent<Object, Object> getCustomerRecord(Integer id, String operation, Long epoch) {
-    return this.getCustomerRecord(id, operation, TestUtil.randomString(12), epoch);
-  }
-
+  
 }
diff --git a/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerUpsertTest.java b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerUpsertTest.java
index 6e49dee6..ea9bb25c 100644
--- a/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerUpsertTest.java
+++ b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerUpsertTest.java
@@ -8,13 +8,11 @@
 
 package io.debezium.server.iceberg;
 
-import io.debezium.engine.ChangeEvent;
 import io.debezium.server.iceberg.testresources.*;
 import io.quarkus.test.common.QuarkusTestResource;
 import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.junit.TestProfile;
 
-import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import javax.inject.Inject;
@@ -40,12 +38,12 @@ public class IcebergChangeConsumerUpsertTest extends BaseSparkTest {
 
   @Test
   public void testSimpleUpsert() throws Exception {
-
+    String dest = "inventory.customers_upsert";
     // test simple inserts
-    List<ChangeEvent<Object, Object>> records = new ArrayList<>();
-    records.add(getCustomerRecord(1, "c"));
-    records.add(getCustomerRecord(2, "c"));
-    records.add(getCustomerRecord(3, "c"));
+    List<io.debezium.engine.ChangeEvent<Object, Object>> records = new ArrayList<>();
+    records.add(TestChangeEvent.of(dest, 1, "c"));
+    records.add(TestChangeEvent.of(dest, 2, "c"));
+    records.add(TestChangeEvent.of(dest, 3, "c"));
     consumer.handleBatch(records, TestUtil.getCommitter());
 
     Dataset<Row> ds = getTableData("testc.inventory.customers_upsert");
@@ -54,10 +52,10 @@ public void testSimpleUpsert() throws Exception {
 
     // 3 records should be updated 4th one should be inserted
     records.clear();
-    records.add(getCustomerRecord(1, "r"));
-    records.add(getCustomerRecord(2, "d"));
-    records.add(getCustomerRecord(3, "u", "UpdatednameV1"));
-    records.add(getCustomerRecord(4, "c"));
+    records.add(TestChangeEvent.of(dest, 1, "r"));
+    records.add(TestChangeEvent.of(dest, 2, "d"));
+    records.add(TestChangeEvent.of(dest, 3, "u", "UpdatednameV1"));
+    records.add(TestChangeEvent.of(dest, 4, "c"));
     consumer.handleBatch(records, TestUtil.getCommitter());
 
     ds = getTableData("testc.inventory.customers_upsert");
@@ -71,17 +69,17 @@ public void testSimpleUpsert() throws Exception {
 
     records.clear();
     // incase of duplicate records it should only keep the latest by epoch ts
-    records.add(getCustomerRecord(3, "r", "UpdatednameV2", 1L));
-    records.add(getCustomerRecord(3, "u", "UpdatednameV3", 2L));
-    records.add(getCustomerRecord(3, "u", "UpdatednameV4", 3L));
-    records.add(getCustomerRecord(4, "u", "Updatedname-4-V1", 4L));
-    records.add(getCustomerRecord(4, "u", "Updatedname-4-V2", 5L));
-    records.add(getCustomerRecord(4, "r", "Updatedname-4-V3", 6L));
-    records.add(getCustomerRecord(5, "r", 7L));
-    records.add(getCustomerRecord(6, "r", 8L));
-    records.add(getCustomerRecord(6, "r", 9L));
-    records.add(getCustomerRecord(6, "u", 10L));
-    records.add(getCustomerRecord(6, "u", "Updatedname-6-V1", 11L));
+    records.add(TestChangeEvent.of(dest, 3, "r", "UpdatednameV2", 1L));
+    records.add(TestChangeEvent.of(dest, 3, "u", "UpdatednameV3", 2L));
+    records.add(TestChangeEvent.of(dest, 3, "u", "UpdatednameV4", 3L));
+    records.add(TestChangeEvent.of(dest, 4, "u", "Updatedname-4-V1", 4L));
+    records.add(TestChangeEvent.of(dest, 4, "u", "Updatedname-4-V2", 5L));
+    records.add(TestChangeEvent.of(dest, 4, "r", "Updatedname-4-V3", 6L));
+    records.add(TestChangeEvent.of(dest, 5, "r", 7L));
+    records.add(TestChangeEvent.of(dest, 6, "r", 8L));
+    records.add(TestChangeEvent.of(dest, 6, "r", 9L));
+    records.add(TestChangeEvent.of(dest, 6, "u", 10L));
+    records.add(TestChangeEvent.of(dest, 6, "u", "Updatedname-6-V1", 11L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     ds = getTableData("testc.inventory.customers_upsert");
     ds.sort("id").show(false);
@@ -94,10 +92,10 @@ public void testSimpleUpsert() throws Exception {
     // in case of duplicate records including epoch ts, its should keep latest one based on operation priority
     // ("c", 1, "r", 2, "u", 3, "d", 4);
     records.clear();
-    records.add(getCustomerRecord(3, "d", "UpdatednameV5", 1L));
-    records.add(getCustomerRecord(3, "u", "UpdatednameV6", 1L));
-    records.add(getCustomerRecord(6, "c", "Updatedname-6-V2", 1L));
-    records.add(getCustomerRecord(6, "r", "Updatedname-6-V3", 1L));
+    records.add(TestChangeEvent.of(dest, 3, "d", "UpdatednameV5", 1L));
+    records.add(TestChangeEvent.of(dest, 3, "u", "UpdatednameV6", 1L));
+    records.add(TestChangeEvent.of(dest, 6, "c", "Updatedname-6-V2", 1L));
+    records.add(TestChangeEvent.of(dest, 6, "r", "Updatedname-6-V3", 1L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     ds = getTableData("testc.inventory.customers_upsert");
     ds.show();
@@ -106,10 +104,10 @@ public void testSimpleUpsert() throws Exception {
 
     // if its not standard insert followed by update! should keep latest one
     records.clear();
-    records.add(getCustomerRecord(7, "u", 1L));
-    records.add(getCustomerRecord(7, "u", 2L));
-    records.add(getCustomerRecord(7, "r", 3L));
-    records.add(getCustomerRecord(7, "u", "Updatedname-7-V1", 4L));
+    records.add(TestChangeEvent.of(dest, 7, "u", 1L));
+    records.add(TestChangeEvent.of(dest, 7, "u", 2L));
+    records.add(TestChangeEvent.of(dest, 7, "r", 3L));
+    records.add(TestChangeEvent.of(dest, 7, "u", "Updatedname-7-V1", 4L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     ds = getTableData("testc.inventory.customers_upsert");
     ds.show();
@@ -119,12 +117,13 @@ public void testSimpleUpsert() throws Exception {
 
   @Test
   public void testSimpleUpsertCompositeKey() throws Exception {
+    String dest = "inventory.customers_upsert_compositekey";
     // test simple inserts
-    List<ChangeEvent<Object, Object>> records = new ArrayList<>();
-    records.add(getCustomerRecordCompositeKey(1, "c", "user1", 1L));
-    records.add(getCustomerRecordCompositeKey(1, "c", "user2", 1L));
-    records.add(getCustomerRecordCompositeKey(1, "u", "user1", 2L));
-    records.add(getCustomerRecordCompositeKey(1, "r", "user1", 3L));
+    List<io.debezium.engine.ChangeEvent<Object, Object>> records = new ArrayList<>();
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "c", "user1", 1L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "c", "user2", 1L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "u", "user1", 2L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "r", "user1", 3L));
     consumer.handleBatch(records, TestUtil.getCommitter());
 
     Dataset<Row> ds = getTableData("testc.inventory.customers_upsert_compositekey");
@@ -133,7 +132,7 @@ public void testSimpleUpsertCompositeKey() throws Exception {
     Assertions.assertEquals(ds.where("id = 1").count(), 2);
 
     records.clear();
-    records.add(getCustomerRecordCompositeKey(1, "u", "user2", 1L));
+    records.add(TestChangeEvent.ofCompositeKey(dest, 1, "u", "user2", 1L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     ds = getTableData("testc.inventory.customers_upsert_compositekey");
     ds.show();
@@ -143,11 +142,12 @@ public void testSimpleUpsertCompositeKey() throws Exception {
 
   @Test
   public void testSimpleUpsertNoKey() throws Exception {
+    String dest = "inventory.customers_upsert_nokey";
     // when there is no PK it should fall back to append mode
-    List<ChangeEvent<Object, Object>> records = new ArrayList<>();
-    records.add(getCustomerRecordNoKey(1, "c", "user1", 1L));
-    records.add(getCustomerRecordNoKey(1, "c", "user2", 1L));
-    records.add(getCustomerRecordNoKey(1, "u", "user1", 2L));
+    List<io.debezium.engine.ChangeEvent<Object, Object>> records = new ArrayList<>();
+    records.add(TestChangeEvent.ofNoKey(dest, 1, "c", "user1", 1L));
+    records.add(TestChangeEvent.ofNoKey(dest, 1, "c", "user2", 1L));
+    records.add(TestChangeEvent.ofNoKey(dest, 1, "u", "user1", 2L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     Dataset<Row> ds = getTableData("testc.inventory.customers_upsert_nokey");
     ds.show();
@@ -155,9 +155,9 @@ public void testSimpleUpsertNoKey() throws Exception {
     Assertions.assertEquals(ds.where("id = 1").count(), 3);
 
     records.clear();
-    records.add(getCustomerRecordNoKey(1, "c", "user2", 1L));
-    records.add(getCustomerRecordNoKey(1, "u", "user2", 1L));
-    records.add(getCustomerRecordNoKey(1, "r", "user1", 3L));
+    records.add(TestChangeEvent.ofNoKey(dest, 1, "c", "user2", 1L));
+    records.add(TestChangeEvent.ofNoKey(dest, 1, "u", "user2", 1L));
+    records.add(TestChangeEvent.ofNoKey(dest, 1, "r", "user1", 3L));
     consumer.handleBatch(records, TestUtil.getCommitter());
     ds = getTableData("testc.inventory.customers_upsert_nokey");
     ds.show();
@@ -165,72 +165,4 @@ public void testSimpleUpsertNoKey() throws Exception {
     Assertions.assertEquals(ds.where("id = 1 AND __op= 'c' AND first_name= 'user2'").count(), 2);
   }
 
-  private TestChangeEvent<Object, Object> getCustomerRecord(Integer id, String operation, String name, Long epoch) {
-    String key = "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false," + "\"field\":\"id\"}]," +
-                 "\"optional\":false,\"name\":\"testc.inventory.customers.Key\"}," +
-                 "\"payload\":{\"id\":" + id + "}}";
-    String val = "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"field\":\"id\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"first_name\"},{\"type\":\"string\",\"optional\":false,\"field\":\"last_name\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"email\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__op\"}," +
-                 "{\"type\":\"string\",\"optional\":true,\"field\":\"__table\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"__lsn\"}," +
-                 "{\"type\":\"int64\",\"optional\":true,\"field\":\"__source_ts_ms\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__deleted\"}]," +
-                 "\"optional\":false,\"name\":\"testc.inventory.customers.Value\"}," +
-                 "\"payload\":{\"id\":" + id + ",\"first_name\":\"" + name + "\",\"last_name\":\"Walker\",\"email\":\"ed@walker" +
-                 ".com\"," +
-                 "\"__op\":\"" + operation + "\",\"__table\":\"customers\",\"__lsn\":33832960,\"__source_ts_ms\":" + epoch + "," +
-                 "\"__deleted\":\"" + operation.equals("d") + "\"}} ";
-    return new TestChangeEvent<>(key, val, "testc.inventory.customers_upsert");
-  }
-
-  private TestChangeEvent<Object, Object> getCustomerRecordCompositeKey(Integer id, String operation, String name,
-                                                                        Long epoch) {
-    String key = "{\"schema\":{\"type\":\"struct\",\"fields\":[" +
-                 "{\"type\":\"int32\",\"optional\":false," + "\"field\":\"id\"}," +
-                 "{\"type\":\"string\",\"optional\":false," + "\"field\":\"first_name\"}" +
-                 "]," +
-                 "\"optional\":false,\"name\":\"testc.inventory.customers.Key\"}," +
-                 "\"payload\":{\"id\":" + id + ",\"first_name\":\"" + name + "\"}}";
-
-    String val = "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"field\":\"id\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"first_name\"},{\"type\":\"string\",\"optional\":false,\"field\":\"last_name\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"email\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__op\"}," +
-                 "{\"type\":\"string\",\"optional\":true,\"field\":\"__table\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"__lsn\"}," +
-                 "{\"type\":\"int64\",\"optional\":true,\"field\":\"__source_ts_ms\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__deleted\"}]," +
-                 "\"optional\":false,\"name\":\"testc.inventory.customers.Value\"}," +
-                 "\"payload\":{\"id\":" + id + ",\"first_name\":\"" + name + "\",\"last_name\":\"Walker\",\"email\":\"ed@walker" +
-                 ".com\"," +
-                 "\"__op\":\"" + operation + "\",\"__table\":\"customers\",\"__lsn\":33832960,\"__source_ts_ms\":" + epoch + "," +
-                 "\"__deleted\":\"" + operation.equals("d") + "\"}} ";
-    return new TestChangeEvent<>(key, val, "testc.inventory.customers_upsert_compositekey");
-  }
-
-  private TestChangeEvent<Object, Object> getCustomerRecordNoKey(Integer id, String operation, String name,
-                                                                 Long epoch) {
-
-    String val = "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"field\":\"id\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"first_name\"},{\"type\":\"string\",\"optional\":false,\"field\":\"last_name\"}," +
-                 "{\"type\":\"string\",\"optional\":false,\"field\":\"email\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__op\"}," +
-                 "{\"type\":\"string\",\"optional\":true,\"field\":\"__table\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"__lsn\"}," +
-                 "{\"type\":\"int64\",\"optional\":true,\"field\":\"__source_ts_ms\"},{\"type\":\"string\",\"optional\":true,\"field\":\"__deleted\"}]," +
-                 "\"optional\":false,\"name\":\"testc.inventory.customers.Value\"}," +
-                 "\"payload\":{\"id\":" + id + ",\"first_name\":\"" + name + "\",\"last_name\":\"Walker\",\"email\":\"ed@walker" +
-                 ".com\"," +
-                 "\"__op\":\"" + operation + "\",\"__table\":\"customers\",\"__lsn\":33832960,\"__source_ts_ms\":" + epoch + "," +
-                 "\"__deleted\":\"" + operation.equals("d") + "\"}} ";
-    return new TestChangeEvent<>(null, val, "testc.inventory.customers_upsert_nokey");
-  }
-
-  private TestChangeEvent<Object, Object> getCustomerRecord(Integer id, String operation) {
-    return this.getCustomerRecord(id, operation, TestUtil.randomString(12), Instant.now().toEpochMilli());
-  }
-
-
-  private TestChangeEvent<Object, Object> getCustomerRecord(Integer id, String operation, String name) {
-    return this.getCustomerRecord(id, operation, name, Instant.now().toEpochMilli());
-  }
-
-  private TestChangeEvent<Object, Object> getCustomerRecord(Integer id, String operation, Long epoch) {
-    return this.getCustomerRecord(id, operation, TestUtil.randomString(12), epoch);
-  }
-
 }
diff --git a/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/tableoperator/IcebergChangeEventBuilderTest.java b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/tableoperator/IcebergChangeEventBuilderTest.java
new file mode 100644
index 00000000..4398084b
--- /dev/null
+++ b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/tableoperator/IcebergChangeEventBuilderTest.java
@@ -0,0 +1,75 @@
+/*
+ *
+ *  * Copyright memiiso Authors.
+ *  *
+ *  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ */
+
+package io.debezium.server.iceberg.tableoperator;
+
+import io.debezium.server.iceberg.IcebergChangeEvent;
+import io.debezium.server.iceberg.testresources.IcebergChangeEventBuilder;
+
+import java.util.List;
+import java.util.Set;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+
+/**
+ * @author Ismail Simsek
+ */
+class IcebergChangeEventBuilderTest {
+
+  @Test
+  public void testIcebergChangeEventBuilder() {
+    Schema schema1 = new Schema(
+        List.of(
+            required(1, "id", Types.IntegerType.get()),
+            optional(2, "data", Types.StringType.get()),
+            optional(3, "preferences", Types.StructType.of(
+                optional(4, "feature1", Types.BooleanType.get()),
+                optional(5, "feature2", Types.BooleanType.get())
+            )),
+            optional(6, "__source_ts", Types.TimestampType.withZone())
+        )
+        , Set.of(1)
+    );
+
+    IcebergChangeEventBuilder b = new IcebergChangeEventBuilder();
+    IcebergChangeEvent t = b.
+        addKeyField("id", 1)
+        .addField("data", "testdatavalue")
+        .addField("preferences", "feature1", true)
+        .addField("preferences", "feature2", true)
+        .build();
+    Assertions.assertTrue(schema1.sameSchema(t.icebergSchema()));
+
+    Schema schema2 = new Schema(
+        optional(1, "id", Types.IntegerType.get()),
+        optional(2, "data", Types.StringType.get()),
+        optional(3, "preferences", Types.StructType.of(
+            optional(4, "feature1", Types.BooleanType.get()),
+            optional(5, "feature2", Types.BooleanType.get())
+        )),
+        optional(6, "__source_ts", Types.TimestampType.withZone())
+    );
+
+    b = new IcebergChangeEventBuilder();
+    t = b.
+        addField("id", 1)
+        .addField("data", "testdatavalue")
+        .addField("preferences", "feature1", true)
+        .addField("preferences", "feature2", true)
+        .build();
+    Assertions.assertTrue(schema2.sameSchema(t.icebergSchema()));
+  }
+
+
+}
\ No newline at end of file
diff --git a/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/IcebergChangeEventBuilder.java b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/IcebergChangeEventBuilder.java
new file mode 100644
index 00000000..75998515
--- /dev/null
+++ b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/IcebergChangeEventBuilder.java
@@ -0,0 +1,171 @@
+/*
+ *
+ *  * Copyright memiiso Authors.
+ *  *
+ *  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ */
+
+package io.debezium.server.iceberg.testresources;
+
+import io.debezium.server.iceberg.IcebergChangeConsumerTest;
+import io.debezium.server.iceberg.IcebergChangeEvent;
+
+import java.util.Iterator;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * helper class used to generate test change events
+ *
+ * @author Ismail Simsek
+ */
+public class IcebergChangeEventBuilder {
+
+  protected static final Logger LOGGER = LoggerFactory.getLogger(IcebergChangeConsumerTest.class);
+  ObjectNode payload = JsonNodeFactory.instance.objectNode();
+  ObjectNode keyPayload = JsonNodeFactory.instance.objectNode();
+  String destination = "test";
+
+  public IcebergChangeEventBuilder() {
+  }
+
+  public IcebergChangeEventBuilder destination(String destination) {
+    this.destination = destination;
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addField(String parentFieldName, String name, String val) {
+    ObjectNode nestedField = JsonNodeFactory.instance.objectNode();
+    nestedField.put(name, val);
+    this.payload.put(parentFieldName, nestedField);
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addField(String parentFieldName, String name, int val) {
+    ObjectNode nestedField = JsonNodeFactory.instance.objectNode();
+    nestedField.put(name, val);
+    this.payload.put(parentFieldName, nestedField);
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addField(String parentFieldName, String name, boolean val) {
+
+    ObjectNode nestedField = JsonNodeFactory.instance.objectNode();
+    if (this.payload.has(parentFieldName)) {
+      nestedField = (ObjectNode) this.payload.get(parentFieldName);
+    }
+    nestedField.put(name, val);
+    this.payload.put(parentFieldName, nestedField);
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addField(String name, int val) {
+    payload.put(name, val);
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addField(String name, String val) {
+    payload.put(name, val);
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addField(String name, long val) {
+    payload.put(name, val);
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addField(String name, double val) {
+    payload.put(name, val);
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addField(String name, boolean val) {
+    payload.put(name, val);
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addKeyField(String name, int val) {
+    keyPayload.put(name, val);
+    payload.put(name, val);
+    return this;
+  }
+
+  public IcebergChangeEventBuilder addKeyField(String name, String val) {
+    keyPayload.put(name, val);
+    payload.put(name, val);
+    return this;
+  }
+
+  public IcebergChangeEvent build() {
+    return new IcebergChangeEvent(
+        this.destination,
+        payload,
+        keyPayload,
+        this.valueSchema(),
+        this.keySchema()
+    );
+  }
+
+  private ObjectNode valueSchema() {
+    return getSchema(payload);
+  }
+
+  private ObjectNode keySchema() {
+    return getSchema(keyPayload);
+  }
+
+  private ObjectNode getSchema(ObjectNode node) {
+    ObjectNode schema = JsonNodeFactory.instance.objectNode();
+
+    ArrayNode fs = getSchemaFields(node);
+    if (fs.isEmpty()) {
+      return null;
+    } else {
+      schema.put("type", "struct");
+      schema.set("fields", fs);
+      return schema;
+    }
+  }
+
+  private ArrayNode getSchemaFields(ObjectNode node) {
+    ArrayNode fields = JsonNodeFactory.instance.arrayNode();
+    Iterator<Map.Entry<String, JsonNode>> iter = node.fields();
+    while (iter.hasNext()) {
+      Map.Entry<String, JsonNode> field = iter.next();
+
+      ObjectNode schemaField = JsonNodeFactory.instance.objectNode();
+      if (field.getValue().isContainerNode()) {
+        schemaField.put("type", "struct");
+        schemaField.set("fields", getSchemaFields((ObjectNode) field.getValue()));
+      } else if (field.getValue().isInt()) {
+        schemaField.put("type", "int32");
+      } else if (field.getValue().isLong()) {
+        schemaField.put("type", "int64");
+      } else if (field.getValue().isBoolean()) {
+        schemaField.put("type", "boolean");
+      } else if (field.getValue().isTextual()) {
+        schemaField.put("type", "string");
+      } else if (field.getValue().isFloat()) {
+        schemaField.put("type", "float64");
+      }
+      if (keyPayload.has(field.getKey())) {
+        schemaField.put("optional", false);
+      } else {
+        schemaField.put("optional", true);
+      }
+      schemaField.put("field", field.getKey());
+      fields.add(schemaField);
+    }
+
+    return fields;
+  }
+
+
+}
\ No newline at end of file
diff --git a/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/TestChangeEvent.java b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/TestChangeEvent.java
index eb20ab67..1fdbf734 100644
--- a/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/TestChangeEvent.java
+++ b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/TestChangeEvent.java
@@ -10,7 +10,15 @@
 
 import io.debezium.engine.ChangeEvent;
 import io.debezium.engine.RecordChangeEvent;
+import io.debezium.server.iceberg.IcebergChangeEvent;
 
+import java.time.Instant;
+
+/**
+ * helper class used to generate test customer change events
+ *
+ * @author Ismail Simsek
+ */
 public class TestChangeEvent<K, V> implements ChangeEvent<K, V>, RecordChangeEvent<V> {
 
   private final K key;
@@ -27,6 +35,81 @@ public TestChangeEvent(V value) {
     this(null, value, null);
   }
 
+  public static TestChangeEvent<Object, Object> of(String destination, Integer id, String operation, String name,
+                                                   Long epoch) {
+    final IcebergChangeEvent t = new IcebergChangeEventBuilder()
+        .destination(destination)
+        .addKeyField("id", id)
+        .addField("first_name", name)
+        .addField("__op", operation)
+        .addField("__source_ts_ms", epoch)
+        .addField("__deleted", operation.equals("d"))
+        .build();
+
+    final String key = "{" +
+                       "\"schema\":" + t.jsonSchema().keySchema() + "," +
+                       "\"payload\":" + t.key() +
+                       "} ";
+    final String val = "{" +
+                       "\"schema\":" + t.jsonSchema().valueSchema() + "," +
+                       "\"payload\":" + t.value() +
+                       "} ";
+    return new TestChangeEvent<>(key, val, destination);
+  }
+
+  public static TestChangeEvent<Object, Object> ofCompositeKey(String destination, Integer id, String operation, String name,
+                                                               Long epoch) {
+    final IcebergChangeEvent t = new IcebergChangeEventBuilder()
+        .destination(destination)
+        .addKeyField("id", id)
+        .addKeyField("first_name", name)
+        .addField("__op", operation)
+        .addField("__source_ts_ms", epoch)
+        .addField("__deleted", operation.equals("d"))
+        .build();
+
+    final String key = "{" +
+                       "\"schema\":" + t.jsonSchema().keySchema() + "," +
+                       "\"payload\":" + t.key() +
+                       "} ";
+    final String val = "{" +
+                       "\"schema\":" + t.jsonSchema().valueSchema() + "," +
+                       "\"payload\":" + t.value() +
+                       "} ";
+
+    return new TestChangeEvent<>(key, val, destination);
+  }
+
+  public static TestChangeEvent<Object, Object> of(String destination, Integer id, String operation) {
+    return of(destination, id, operation, TestUtil.randomString(12), Instant.now().toEpochMilli());
+  }
+
+  public static TestChangeEvent<Object, Object> of(String destination, Integer id, String operation, String name) {
+    return of(destination, id, operation, name, Instant.now().toEpochMilli());
+  }
+
+  public static TestChangeEvent<Object, Object> of(String destination, Integer id, String operation, Long epoch) {
+    return of(destination, id, operation, TestUtil.randomString(12), epoch);
+  }
+
+  public static TestChangeEvent<Object, Object> ofNoKey(String destination, Integer id, String operation, String name,
+                                                        Long epoch) {
+    final IcebergChangeEvent t = new IcebergChangeEventBuilder()
+        .destination(destination)
+        .addField("id", id)
+        .addField("first_name", name)
+        .addField("__op", operation)
+        .addField("__source_ts_ms", epoch)
+        .addField("__deleted", operation.equals("d"))
+        .build();
+
+    final String val = "{" +
+                       "\"schema\":" + t.jsonSchema().valueSchema() + "," +
+                       "\"payload\":" + t.value() +
+                       "} ";
+    return new TestChangeEvent<>(null, val, destination);
+  }
+
   @Override
   public K key() {
     return key;
@@ -51,4 +134,5 @@ public String destination() {
   public String toString() {
     return "EmbeddedEngineChangeEvent [key=" + key + ", value=" + value + ", sourceRecord=" + destination + "]";
   }
+
 }