danielcweeks commented on code in PR #15498:
URL: https://github.com/apache/iceberg/pull/15498#discussion_r2921175325
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -531,4 +537,62 @@ private String ensureTimestampFormat(String str) {
}
return result;
}
+
+ private Variant tryDeserializeVariant(Object value) {
+ // Try to deserialize ByteBuffer or byte[] as a serialized variant
+ // Returns null if not a valid serialized variant
+ try {
+ if (value instanceof ByteBuffer) {
+ return Variant.from((ByteBuffer) value);
+ } else if (value instanceof byte[]) {
+ return Variant.from(ByteBuffer.wrap((byte[]) value));
+ }
+ } catch (RuntimeException e) {
+ // Not a valid serialized variant, return null to fall back to JSON
conversion
+ }
+ return null;
+ }
+
+ protected Variant convertVariant(Object value) {
+ try {
+ // If it's already a Variant, return it
+ if (value instanceof Variant) {
+ return (Variant) value;
+ }
+
+ // Try to deserialize if it's a ByteBuffer or byte[]
+ Variant deserialized = tryDeserializeVariant(value);
Review Comment:
I don't think this is a safe way to handle binary data coming through. We
should assume binary data is just bytes. If we "try" to process a bytebuffer
we may affect the internal state and not all implementations may support
mark/reset behavior. It appears that we're also dropping it if we can't
deserialize it, which isn't good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
