Skip to content

Kafka Connect: Pre-size collections in RecordConverter list/map conversion #16657

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
laskoviymishka merged 3 commits into from
Jun 18, 2026
+32 −15
Merged

Kafka Connect: Pre-size collections in RecordConverter list/map conversion #16657

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a1a4220
Kafka Connect: Pre-size collections in RecordConverter list/map conve…
Jun 2, 2026
d844580
Trigger CI
Jun 12, 2026
c0c3390
Kafka Connect: Add empty list/map conversion test for RecordConverter
Jun 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 17 additions & 15 deletions ...-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
import org.apache.iceberg.mapping.NameMapping;
import org.apache.iceberg.mapping.NameMappingParser;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.Type;
Expand Down Expand Up @@ -300,28 +301,29 @@ protected List<Object> convertListValue(
Object value, ListType type, SchemaUpdate.Consumer schemaUpdateConsumer) {
Preconditions.checkArgument(value instanceof List);
List<?> list = (List<?>) value;
return list.stream()
.map(
element -> {
int fieldId = type.fields().get(0).fieldId();
return convertValue(element, type.elementType(), fieldId, schemaUpdateConsumer);
})
.collect(Collectors.toList());
int elementFieldId = type.fields().get(0).fieldId();
Type elementType = type.elementType();
List<Object> result = Lists.newArrayListWithCapacity(list.size());

@laskoviymishka laskoviymishka Jun 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

While we're here — TestRecordConverter only ever runs these through 2-element lists/maps, so the new zero-capacity path (newArrayListWithCapacity(0) / newHashMapWithExpectedSize(0)) isn't exercised at all.

I'd add one test that converts a record with an empty list field and an empty map field and asserts empty results. Cheap, and it closes a pre-existing gap rather than just papering over this change.

@wombatu-kun wombatu-kun Jun 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done c0c3390

for (Object element : list) {
result.add(convertValue(element, elementType, elementFieldId, schemaUpdateConsumer));
}
return result;
}

protected Map<Object, Object> convertMapValue(
Object value, MapType type, SchemaUpdate.Consumer schemaUpdateConsumer) {
Preconditions.checkArgument(value instanceof Map);
Map<?, ?> map = (Map<?, ?>) value;

@laskoviymishka laskoviymishka Jun 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor semantic shift: get(1) used to be evaluated lazily inside the lambda, so an empty map never touched it — now it's eager at method entry. MapType always has exactly two fields so there's no real regression, just worth being aware of. No change needed.

@wombatu-kun wombatu-kun Jun 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The empty-map test added in c0c3390 runs this method with size 0, so the now-eager get(1) is exercised on the empty path you flagged.

Map<Object, Object> result = Maps.newHashMap();
int keyFieldId = type.fields().get(0).fieldId();
int valueFieldId = type.fields().get(1).fieldId();
Type keyType = type.keyType();
Type valueType = type.valueType();
Map<Object, Object> result = Maps.newHashMapWithExpectedSize(map.size());
map.forEach(
(k, v) -> {
int keyFieldId = type.fields().get(0).fieldId();
int valueFieldId = type.fields().get(1).fieldId();
result.put(
convertValue(k, type.keyType(), keyFieldId, schemaUpdateConsumer),
convertValue(v, type.valueType(), valueFieldId, schemaUpdateConsumer));
});
(k, v) ->
result.put(
convertValue(k, keyType, keyFieldId, schemaUpdateConsumer),
convertValue(v, valueType, valueFieldId, schemaUpdateConsumer)));
return result;
}

Expand Down
15 changes: 15 additions & 0 deletions ...nect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,21 @@ public void testMapConvert() {
assertRecordValues(record);
}

@Test
public void testEmptyListAndMapConvert() {
Table table = mock(Table.class);
when(table.schema()).thenReturn(SCHEMA);
RecordConverter converter = new RecordConverter(table, config);

Map<String, Object> data = Maps.newHashMap(createMapData());
data.put("li", ImmutableList.of());
data.put("ma", ImmutableMap.of());

Record record = converter.convert(data);
assertThat((List<?>) record.getField("li")).isEmpty();
assertThat((Map<?, ?>) record.getField("ma")).isEmpty();
}

@Test
public void testUUIDConversionWithParquet() {
Table table = mock(Table.class);
Expand Down