Skip to content

[KafkaConnect] Fix RecordConverter for UUID and Fixed Types #11346

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
RussellSpitzer merged 10 commits into from
Oct 25, 2024
Merged

[KafkaConnect] Fix RecordConverter for UUID and Fixed Types #11346

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
84730e5
[KafkaConnect] Fix RecordConverter
Oct 17, 2024
718f62c
update E2E test
Oct 18, 2024
6e2cb73
change test
Oct 18, 2024
f964819
make all the formats pass
Oct 18, 2024
67fcc0f
remove orc test for now
Oct 18, 2024
4e876ac
add more tests and revert todo
Oct 22, 2024
83e5393
Address review feedback
Oct 22, 2024
715269e
fix missed part of code
Oct 23, 2024
9e6bb4b
Add parquet uuid test
Oct 24, 2024
aaa8b33
remove dependency
Oct 25, 2024
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
26 changes: 21 additions & 5 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 @@ -37,9 +37,11 @@
import java.util.Base64;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Collectors;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
Expand All @@ -59,7 +61,9 @@
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.types.Types.StructType;
import org.apache.iceberg.types.Types.TimestampType;
import org.apache.iceberg.util.ByteBuffers;
import org.apache.iceberg.util.DateTimeUtil;
import org.apache.iceberg.util.UUIDUtil;
import org.apache.kafka.connect.data.Struct;

class RecordConverter {
Expand Down Expand Up @@ -128,8 +132,9 @@ private Object convertValue(
case UUID:
return convertUUID(value);
case BINARY:
case FIXED:
return convertBase64Binary(value);
case FIXED:
return ByteBuffers.toByteArray(convertBase64Binary(value));
case DATE:
return convertDateValue(value);
case TIME:
Expand Down Expand Up @@ -388,13 +393,24 @@ protected String convertString(Object value) {
throw new IllegalArgumentException("Cannot convert to string: " + value.getClass().getName());
}

protected UUID convertUUID(Object value) {
protected Object convertUUID(Object value) {
UUID uuid;
if (value instanceof String) {
return UUID.fromString((String) value);
uuid = UUID.fromString((String) value);
} else if (value instanceof UUID) {
return (UUID) value;
uuid = (UUID) value;
} else {
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))) {
return UUIDUtil.convert(uuid);
} else {
return uuid;
}
throw new IllegalArgumentException("Cannot convert to UUID: " + value.getClass().getName());
}

protected ByteBuffer convertBase64Binary(Object value) {
Expand Down
32 changes: 27 additions & 5 deletions ...nect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,9 +37,11 @@
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.UUID;
import java.util.function.Function;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.connect.IcebergSinkConfig;
Expand Down Expand Up @@ -72,6 +74,7 @@
import org.apache.iceberg.types.Types.TimeType;
import org.apache.iceberg.types.Types.TimestampType;
import org.apache.iceberg.types.Types.UUIDType;
import org.apache.iceberg.util.UUIDUtil;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
Expand Down Expand Up @@ -221,6 +224,25 @@ public void testMapConvert() {
assertRecordValues(record);
}

@Test
public void testUUIDConversionWithParquet() {
Table table = mock(Table.class);
when(table.schema())
.thenReturn(new org.apache.iceberg.Schema(NestedField.required(1, "uuid", UUIDType.get())));
when(config.writeProps())
.thenReturn(
ImmutableMap.of(
TableProperties.DEFAULT_FILE_FORMAT,
FileFormat.PARQUET.name().toLowerCase(Locale.ROOT)));

RecordConverter converter = new RecordConverter(table, config);
Map<String, Object> data =
ImmutableMap.<String, Object>builder().put("uuid", UUID_VAL.toString()).build();

Record record = converter.convert(data);
assertThat(record.getField("uuid")).isEqualTo(UUIDUtil.convert(UUID_VAL));
}

@Test
public void testNestedMapConvert() {
Table table = mock(Table.class);
Expand Down Expand Up @@ -859,7 +881,7 @@ public void testEvolveTypeDetectionStructNested() {
assertThat(updateMap.get("st.ff").type()).isInstanceOf(DoubleType.class);
}

private Map<String, Object> createMapData() {
public static Map<String, Object> createMapData() {
return ImmutableMap.<String, Object>builder()
.put("i", 1)
.put("l", 2L)
Expand Down Expand Up @@ -898,8 +920,8 @@ private Struct createStructData() {
.put("s", STR_VAL)
.put("b", true)
.put("u", UUID_VAL.toString())
.put("f", BYTES_VAL.array())
.put("bi", BYTES_VAL.array())
.put("f", BYTES_VAL)
.put("bi", BYTES_VAL)
.put("li", LIST_VAL)
.put("ma", MAP_VAL);
}
Expand All @@ -921,11 +943,11 @@ private void assertRecordValues(Record record) {
assertThat(rec.getField("dec")).isEqualTo(DEC_VAL);
assertThat(rec.getField("s")).isEqualTo(STR_VAL);
assertThat(rec.getField("b")).isEqualTo(true);
assertThat(rec.getField("u")).isEqualTo(UUID_VAL);
assertThat(rec.getField("f")).isEqualTo(BYTES_VAL);
assertThat(rec.getField("f")).isEqualTo(BYTES_VAL.array());
assertThat(rec.getField("bi")).isEqualTo(BYTES_VAL);
assertThat(rec.getField("li")).isEqualTo(LIST_VAL);
assertThat(rec.getField("ma")).isEqualTo(MAP_VAL);
assertThat(rec.getField("u")).isEqualTo(UUID_VAL);
}

private void assertNestedRecordValues(Record record) {
Expand Down