Skip to content

Kafka Connect: Precompute UUID-as-bytes flag in RecordConverter #16654

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

Closed
wombatu-kun wants to merge 1 commit into from
+11 −4
Closed

Kafka Connect: Precompute UUID-as-bytes flag in RecordConverter #16654

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
13 changes: 9 additions & 4 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 @@ -95,11 +95,19 @@ class RecordConverter {
private final NameMapping nameMapping;
private final IcebergSinkConfig config;
private final Map<Integer, Map<String, NestedField>> structNameMap = Maps.newHashMap();
// Parquet stores UUIDs as a 16-byte fixed; other formats keep the UUID logical type. The write
// file format is fixed for the converter's lifetime, so resolve this once instead of per value.
private final boolean writeUuidAsBytes;

RecordConverter(Table table, IcebergSinkConfig config) {
this.tableSchema = table.schema();
this.nameMapping = createNameMapping(table);
this.config = config;
this.writeUuidAsBytes =

@laskoviymishka laskoviymishka Aug 7, 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.

Reading config.writeProps() here means any RecordConverter built from a bare IcebergSinkConfig mock now NPEs at construction rather than lazily inside convertUUID. That's arguably the healthier failure mode, and you've covered it with the @BeforeEach default — just worth a quick check that no other test in the module constructs a converter with an unstubbed config.

It also freezes the value at construction, so a future test that constructs first and re-stubs writeProps() afterward would silently skip the Parquet branch. Not a problem today, but the ordering constraint is now implicit — a one-line comment on the field would save someone a confusing debug session. wdyt?

@wombatu-kun wombatu-kun Aug 8, 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.

No test was at risk here. IcebergWriter.initNewWriter() calls RecordUtils.createTableWriter before new RecordConverter(...), and that already does tableProps.putAll(config.writeProps()). TestSinkWriter builds a real SinkWriter over a bare mock that never stubs writeProps() and passes on main, because Mockito returns an empty map rather than null for Map-returning methods.

FileFormat.PARQUET
.name()
.toLowerCase(Locale.ROOT)
.equals(config.writeProps().get(TableProperties.DEFAULT_FILE_FORMAT));

@laskoviymishka laskoviymishka Aug 7, 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.

Pre-existing, but since it moved into new code: the comparison is case-sensitive on the property value, so a user who sets write.format.default=PARQUET (or Parquet) gets writeUuidAsBytes=false and the UUID stored as an object instead of bytes. FileFormat.PARQUET.name().equalsIgnoreCase(...) would harden it. Optional, and fine to leave for a follow-up.

@wombatu-kun wombatu-kun Aug 8, 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.

Hardening the comparison would widen a branch that needs removing instead: it has thrown ClassCastException: [B cannot be cast to java.util.UUID since #11904 changed the Parquet UUID writer to take a java.util.UUID. Tracked in #17076, fix in #17079.

}

Record convert(Object data) {
Expand Down Expand Up @@ -420,10 +428,7 @@ protected Object convertUUID(Object value) {
throw new IllegalArgumentException("Cannot convert to UUID: " + value.getClass().getName());
}

if (FileFormat.PARQUET
.name()
.toLowerCase(Locale.ROOT)
.equals(config.writeProps().get(TableProperties.DEFAULT_FILE_FORMAT))) {
if (writeUuidAsBytes) {
return UUIDUtil.convert(uuid);
} else {
return uuid;
Expand Down
2 changes: 2 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 @@ -220,6 +220,8 @@ public static void beforeAll() {
public void before() {
this.config = mock(IcebergSinkConfig.class);
when(config.jsonConverter()).thenReturn(JSON_CONVERTER);
// production writeProps() is never null; default it so RecordConverter construction succeeds
when(config.writeProps()).thenReturn(ImmutableMap.of());
}

@Test
Expand Down