Add iceberg v2 consumer for flattened data consuming, schema will be inferred by arrow

Author: ismailsimsek

README.md

@@ -30,6 +30,7 @@ The Iceberg handlers are designed to stream CDC events directly into Apache Iceb
 *   `IcebergChangeHandler`: A straightforward handler that appends change data to a source-equivalent Iceberg table using a predefined schema.
     *   **Use Case**: Best for creating a "bronze" layer where you want to capture the raw Debezium event. The `before` and `after` payloads are stored as complete JSON strings.
     *   **Schema**: Uses a fixed schema where complex nested fields (`source`, `before`, `after`) are stored as `StringType`. It also includes helpful metadata columns (`_consumed_at`, `_dbz_event_key`, `_dbz_event_key_hash`) for traceability.
+      * With consuming data as json, all source syste schema changes will be absorbed automatically. 
 
 *   `IcebergChangeHandlerV2`: A more advanced handler that automatically infers the schema from the Debezium events and creates a well-structured Iceberg table accordingly.
     *   **Use Case**: Ideal for scenarios where you want the pipeline to automatically create tables with native data types that mirror the source. This allows for direct querying of the data without needing to parse JSON.

tests/test_iceberg_handlerv2.py

@@ -110,10 +110,9 @@ def test_iceberg_handler(self):
         self.execute_on_source_db("DELETE FROM inventory.customers WHERE id = 1002 ;")
         self.execute_on_source_db("ALTER TABLE inventory.customers ADD birth_date date;")
         self.execute_on_source_db("UPDATE inventory.customers SET birth_date = '2020-01-01'  WHERE id = 1001 ;")
-        with self.assertLogs(IcebergChangeHandlerV2.LOGGER_NAME, level='INFO') as cm:
-            # run async then interrupt after timeout time to test the result!
-            Utils.run_engine_async(engine=engine, timeout_sec=44)
-
+        # run
+        Utils.run_engine_async(engine=engine, timeout_sec=44)
+        # test
         data = tbl.scan().to_arrow()
         self.pprint_table(data=data)
         self.assertEqual(data.num_rows, 4)