Kafka Connect: Support VARIANT when record convert by seokyun-ha-toss · Pull Request #15283 · apache/iceberg

seokyun-ha-toss · 2026-02-10T09:11:28Z

Summary

Add support for converting arbitrary Java objects (e.g. Map<String, Object>, lists, primitives) into Iceberg Variant type in the Kafka Connect RecordConverter. Nested maps and lists are converted recursively so that structures like {"user": {"name": "alice", "address": {"city": "Seoul", "zip": "12345"}}} are correctly represented as a single Variant.

Motivation

Kafka Connect payloads often come as schema-less or JSON-like maps. To write them into Iceberg tables with a Variant column, the connector must convert these Java objects into the Variant format (metadata + value) and support nested maps/arrays without losing structure or key names.

Behaviour

Input	Result
Primitives (String, int, long, boolean, etc.)	Single metadata (empty) + corresponding Variant primitive.
Flat map e.g. `{"a": 1, "b": "x"}`	One metadata with keys `["a", "b"]`, one ShreddedObject with two fields.
Nested map e.g. `{"user": {"name": "alice", "address": {"city": "Seoul", "zip": "12345"}}}`	One shared metadata for all keys; root and nested objects as ShreddedObjects with consistent field IDs.
Lists	Converted to VariantArray with elements converted recursively.
Already Variant / ByteBuffer	Pass-through or `Variant.from(ByteBuffer)` where appropriate.

Relates

Variant Data Type Support #10392

Thanks, Good Day!

…ersion methods for nested structures

emkornfield · 2026-02-10T18:34:22Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

    return convertLocalDateTime(value);
  }

+  protected Variant convertVariantValue(Object value) {


Is there a reason this is protected? If it is intended to be overridden it would be good to document the contract that implementors must follow (and what the default implementation does).

I guess other methods in this class don't do that, but seems like maybe we should start?

I followed the existing pattern of the convertXXX() methods, which are all defined as protected. For consistency, I kept the same visibility here.

Please let me know if you think it should be adjusted.

emkornfield · 2026-02-10T18:35:57Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

+    if (value instanceof Map) {
+      Map<?, ?> map = (Map<?, ?>) value;
+      List<String> names = Lists.newArrayList();
+      map.keySet().stream().map(Object::toString).forEach(names::add);


Why not do the iteration once by visiting the entry set below?

This also seems more consistent with the loop below on line 509

emkornfield · 2026-02-10T18:38:21Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

    return convertLocalDateTime(value);
  }

+  protected Variant convertVariantValue(Object value) {


I'm not sure if it is generally done in this project but it might be good to consider having protection against stack overflow on recursion

If we're recursing through the schema/object structure, we don't assume the need to protect against stack overflow (either you can traverse the structure or not). I guess there may be some way to construct an object that will cycle, but that's not typical with how the object mapping works in kafka (it would probably fail upstream of this point anyway).

Good point. However, as @danielcweeks mentioned, actual Kafka values don't have cycles, so the recursion ends at some point.

In my opinion, the stack overflow concern is unavoidable when traversing arbitrarily nested values. It depends on the size and depth of the input data, and in many cases we cannot control it.

Also, some methods (convertValue(), convertStructValue(), convertListValue()) already work recursively.

My concern was more around, very deeply nested values (not necessarily infinite recursion), but if we don't check these in the project, then I don't have concern for not doing so here

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

emkornfield · 2026-02-10T18:48:22Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

+   * Collects all field names (map keys) from the entire object tree. Used to build a single
+   * VariantMetadata for the whole Variant (required for nested maps).
+   */
+  private static List<String> collectFieldNames(Object value) {


it seems like you can maybe avoid some object churn by passing through a Set that gets elements added to as it goes down on recursion?

+1 to @emkornfield's comment. We should avoid navigating the object multiple times.

Great! I'll apply your feedbacks! Thanks!

I'm not sure if two pass is strictly avoidable as it looks like we are trying to sort keys globally?

emkornfield · 2026-02-10T18:53:42Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

+    if (value instanceof byte[]) {
+      return Variants.of(ByteBuffer.wrap((byte[]) value));
+    }
+    if (value instanceof BigDecimal) {


I think this can be folded underneath Number handling?

emkornfield · 2026-02-10T18:55:18Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

+    if (num.doubleValue() == num.longValue()) {
+      return Variants.of(num.longValue());
+    }
+    return Variants.of(num.doubleValue());


It seems safer to throw an exception here, otherwise we potentially have data loss (or at least this should be optional)?

Or we should convert it to BigDecimal.

I agree with failing if it's not a known type. We shouldn't try to massage numeric values if we don't know exactly what is being represented.

emkornfield · 2026-02-10T18:56:53Z

...connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordConverter.java

    assertThat(updateMap.get("st.ff").type()).isInstanceOf(DoubleType.class);
  }

+  @Test


I think it might be more maintainable to write tests against convertVariantValue instead of the static helpers.

Also it seems like there should be tests for convertVariantValue specifically?

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

danielcweeks · 2026-02-10T20:47:14Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

+    if (value instanceof Map) {
+      Map<?, ?> map = (Map<?, ?>) value;
+      List<String> names = Lists.newArrayList();
+      map.keySet().stream().map(Object::toString).forEach(names::add);


Variant keys must be strings, but an arbitrary map passed in through KC does not have that requirement, so we should protect against it.

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

seokyun-ha-toss added 2 commits February 10, 2026 17:17

feat: Implement support for VARIANT type in RecordConverter with conv…

033f47f

…ersion methods for nested structures

test: implement unit test for RecordConverter for Variant

a53af92

github-actions bot added the KAFKACONNECT label Feb 10, 2026

chore: lint spotlessApply

73ea656

emkornfield reviewed Feb 10, 2026

View reviewed changes

danielcweeks reviewed Feb 10, 2026

View reviewed changes

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java Outdated Show resolved Hide resolved

danielcweeks reviewed Feb 10, 2026

View reviewed changes

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java Outdated Show resolved Hide resolved

fix: there's no already Variant value, and use Lists.of()

2b3fcb1

Conversation

seokyun-ha-toss commented Feb 10, 2026

Summary

Motivation

Behaviour

Relates

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants