From fa440732d146448a3990110e354338845dcac267 Mon Sep 17 00:00:00 2001
From: julien-alpaca <165802699+julien-alpaca@users.noreply.github.com>
Date: Wed, 20 Nov 2024 10:37:47 -0500
Subject: [PATCH] [HOTFIX] Convert UUIDs to a bytes array Iceberg store UUIDs
 as 16-byte array: https://iceberg.apache.org/spec/#primitive-types. Returning
 a UUID object throws a ClassCastException. See this issue:
 https://github.com/databricks/iceberg-kafka-connect/issues/314

---
 .../iceberg/connect/data/RecordConverter.java      | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

diff --git a/kafka-connect/src/main/java/io/tabular/iceberg/connect/data/RecordConverter.java b/kafka-connect/src/main/java/io/tabular/iceberg/connect/data/RecordConverter.java
index b62bd1a7..ae8a5a6f 100644
--- a/kafka-connect/src/main/java/io/tabular/iceberg/connect/data/RecordConverter.java
+++ b/kafka-connect/src/main/java/io/tabular/iceberg/connect/data/RecordConverter.java
@@ -384,15 +384,23 @@ protected String convertString(Object value) {
     throw new IllegalArgumentException("Cannot convert to string: " + value.getClass().getName());
   }
 
-  protected UUID convertUUID(Object value) {
+  protected byte[] convertUUID(Object value) {
     if (value instanceof String) {
-      return UUID.fromString((String) value);
+      return asBytes(UUID.fromString((String) value));
     } else if (value instanceof UUID) {
-      return (UUID) value;
+      return asBytes((UUID)value);
     }
     throw new IllegalArgumentException("Cannot convert to UUID: " + value.getClass().getName());
   }
 
+  protected static byte[] asBytes(UUID uuid) {
+    byte[] bytes = new byte[16];
+    java.nio.ByteBuffer.wrap(bytes)
+            .putLong(uuid.getMostSignificantBits())
+            .putLong(uuid.getLeastSignificantBits());
+    return bytes;
+  }
+
   protected ByteBuffer convertBase64Binary(Object value) {
     if (value instanceof String) {
       return ByteBuffer.wrap(Base64.getDecoder().decode((String) value));