This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 561a4e326f Kafka Connect: Split route field path once instead of per
record (#16655)
561a4e326f is described below
commit 561a4e326fe77704e1e3e1d09507f121491a6b54
Author: Vova Kolmakov <[email protected]>
AuthorDate: Mon Oct 12 16:55:53 2026 +0700
Kafka Connect: Split route field path once instead of per record (#16655)
Co-authored-by: Vova Kolmakov <[email protected]>
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.../apache/iceberg/connect/data/RecordUtils.java | 4 +--
.../apache/iceberg/connect/data/SinkWriter.java | 19 +++++++------
.../iceberg/connect/data/TestRecordUtils.java | 32 +++++++++++++++-------
.../iceberg/connect/data/TestSinkWriter.java | 16 +++++++++++
4 files changed, 49 insertions(+), 22 deletions(-)
diff --git
a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordUtils.java
b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordUtils.java
index 3d8f2adf7b..4ad08dd672 100644
---
a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordUtils.java
+++
b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordUtils.java
@@ -35,7 +35,6 @@ import org.apache.iceberg.io.OutputFileFactory;
import org.apache.iceberg.io.TaskWriter;
import org.apache.iceberg.io.UnpartitionedWriter;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.base.Splitter;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.relocated.com.google.common.primitives.Ints;
@@ -49,8 +48,7 @@ class RecordUtils {
@SuppressWarnings("unchecked")
static Object extractFromRecordValue(
- Object recordValue, String fieldName, IcebergSinkConfig config) {
- List<String> fields = Splitter.on('.').splitToList(fieldName);
+ Object recordValue, List<String> fields, IcebergSinkConfig config) {
if (recordValue instanceof Struct) {
return valueFromStruct((Struct) recordValue, fields,
config.replaceNullWithDefault());
} else if (recordValue instanceof Map) {
diff --git
a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java
b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java
index 97e14f8b7e..52ead20f8d 100644
---
a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java
+++
b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java
@@ -30,6 +30,7 @@ import java.util.stream.Collectors;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.connect.IcebergSinkConfig;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.sink.SinkRecord;
@@ -39,12 +40,15 @@ public class SinkWriter {
private final IcebergWriterFactory writerFactory;
private final Map<String, RecordWriter> writers;
private final Map<TopicPartition, Offset> sourceOffsets;
+ private final List<String> routeFieldPath;
public SinkWriter(Catalog catalog, IcebergSinkConfig config) {
this.config = config;
this.writerFactory = new IcebergWriterFactory(catalog, config);
this.writers = Maps.newHashMap();
this.sourceOffsets = Maps.newHashMap();
+ String routeField = config.tablesRouteField();
+ this.routeFieldPath = routeField == null ? null :
Splitter.on('.').splitToList(routeField);
}
public void close() {
@@ -90,9 +94,7 @@ public class SinkWriter {
}
private void routeRecordStatically(SinkRecord record) {
- String routeField = config.tablesRouteField();
-
- if (routeField == null) {
+ if (routeFieldPath == null) {
// route to all tables
config
.tables()
@@ -102,7 +104,7 @@ public class SinkWriter {
});
} else {
- String routeValue = extractRouteValue(record.value(), routeField);
+ String routeValue = extractRouteValue(record.value());
if (routeValue != null) {
config
.tables()
@@ -118,21 +120,20 @@ public class SinkWriter {
}
private void routeRecordDynamically(SinkRecord record) {
- String routeField = config.tablesRouteField();
- Preconditions.checkNotNull(routeField, "Route field cannot be null with
dynamic routing");
+ Preconditions.checkNotNull(routeFieldPath, "Route field cannot be null
with dynamic routing");
- String routeValue = extractRouteValue(record.value(), routeField);
+ String routeValue = extractRouteValue(record.value());
if (routeValue != null) {
String tableName = routeValue.toLowerCase(Locale.ROOT);
writerForTable(tableName, record, true).write(record);
}
}
- private String extractRouteValue(Object recordValue, String routeField) {
+ private String extractRouteValue(Object recordValue) {
if (recordValue == null) {
return null;
}
- Object routeValue = RecordUtils.extractFromRecordValue(recordValue,
routeField, config);
+ Object routeValue = RecordUtils.extractFromRecordValue(recordValue,
routeFieldPath, config);
return routeValue == null ? null : routeValue.toString();
}
diff --git
a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordUtils.java
b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordUtils.java
index 413b90dde4..b8414e1afe 100644
---
a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordUtils.java
+++
b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordUtils.java
@@ -28,6 +28,7 @@ import java.util.UUID;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.connect.IcebergSinkConfig;
import org.apache.iceberg.connect.events.TableReference;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
@@ -51,7 +52,8 @@ public class TestRecordUtils extends WriterTestBase {
public void testExtractFromRecordValueStruct() {
Schema valSchema = SchemaBuilder.struct().field("key",
Schema.INT64_SCHEMA).build();
Struct val = new Struct(valSchema).put("key", 123L);
- Object result = RecordUtils.extractFromRecordValue(val, "key", config);
+ Object result =
+ RecordUtils.extractFromRecordValue(val,
Splitter.on('.').splitToList("key"), config);
assertThat(result).isEqualTo(123L);
}
@@ -65,7 +67,9 @@ public class TestRecordUtils extends WriterTestBase {
Struct data = new Struct(dataSchema).put("id", id);
Struct val = new Struct(valSchema).put("data", data);
- Object result = RecordUtils.extractFromRecordValue(val, "data.id.key",
config);
+ Object result =
+ RecordUtils.extractFromRecordValue(
+ val, Splitter.on('.').splitToList("data.id.key"), config);
assertThat(result).isEqualTo(123L);
}
@@ -74,10 +78,11 @@ public class TestRecordUtils extends WriterTestBase {
Schema valSchema = SchemaBuilder.struct().field("key",
Schema.INT64_SCHEMA).build();
Struct val = new Struct(valSchema).put("key", 123L);
- Object result = RecordUtils.extractFromRecordValue(val, "", config);
+ Object result =
+ RecordUtils.extractFromRecordValue(val,
Splitter.on('.').splitToList(""), config);
assertThat(result).isNull();
- result = RecordUtils.extractFromRecordValue(val, "xkey", config);
+ result = RecordUtils.extractFromRecordValue(val,
Splitter.on('.').splitToList("xkey"), config);
assertThat(result).isNull();
}
@@ -93,7 +98,8 @@ public class TestRecordUtils extends WriterTestBase {
.build();
Struct val = new Struct(valSchema).put("key", null);
- Object result = RecordUtils.extractFromRecordValue(val, "key", config);
+ Object result =
+ RecordUtils.extractFromRecordValue(val,
Splitter.on('.').splitToList("key"), config);
if (replaceNullWithDefault) {
assertThat(result).isEqualTo("default");
} else {
@@ -118,7 +124,9 @@ public class TestRecordUtils extends WriterTestBase {
Struct data = new Struct(dataSchema).put("id", id);
Struct val = new Struct(valSchema).put("data", data);
- Object result = RecordUtils.extractFromRecordValue(val, "data.id.key",
config);
+ Object result =
+ RecordUtils.extractFromRecordValue(
+ val, Splitter.on('.').splitToList("data.id.key"), config);
if (replaceNullWithDefault) {
assertThat(result).isEqualTo(123L);
} else {
@@ -129,7 +137,8 @@ public class TestRecordUtils extends WriterTestBase {
@Test
public void testExtractFromRecordValueMap() {
Map<String, Object> val = ImmutableMap.of("key", 123L);
- Object result = RecordUtils.extractFromRecordValue(val, "key", config);
+ Object result =
+ RecordUtils.extractFromRecordValue(val,
Splitter.on('.').splitToList("key"), config);
assertThat(result).isEqualTo(123L);
}
@@ -139,7 +148,9 @@ public class TestRecordUtils extends WriterTestBase {
Map<String, Object> data = ImmutableMap.of("id", id);
Map<String, Object> val = ImmutableMap.of("data", data);
- Object result = RecordUtils.extractFromRecordValue(val, "data.id.key",
config);
+ Object result =
+ RecordUtils.extractFromRecordValue(
+ val, Splitter.on('.').splitToList("data.id.key"), config);
assertThat(result).isEqualTo(123L);
}
@@ -147,10 +158,11 @@ public class TestRecordUtils extends WriterTestBase {
public void testExtractFromRecordValueMapNull() {
Map<String, Object> val = ImmutableMap.of("key", 123L);
- Object result = RecordUtils.extractFromRecordValue(val, "", config);
+ Object result =
+ RecordUtils.extractFromRecordValue(val,
Splitter.on('.').splitToList(""), config);
assertThat(result).isNull();
- result = RecordUtils.extractFromRecordValue(val, "xkey", config);
+ result = RecordUtils.extractFromRecordValue(val,
Splitter.on('.').splitToList("xkey"), config);
assertThat(result).isNull();
}
diff --git
a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java
b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java
index 40a0673517..af4ec58e4d 100644
---
a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java
+++
b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java
@@ -204,6 +204,22 @@ public class TestSinkWriter {
assertThat(writerResult.tableReference().identifier()).isEqualTo(TABLE_IDENTIFIER);
}
+ @Test
+ void dynamicRouteNestedField() {
+ IcebergSinkConfig config = mock(IcebergSinkConfig.class);
+
when(config.tables()).thenReturn(ImmutableList.of(TABLE_IDENTIFIER.toString()));
+ when(config.tableConfig(any())).thenReturn(mock(TableSinkConfig.class));
+ when(config.dynamicTablesEnabled()).thenReturn(true);
+ when(config.tablesRouteField()).thenReturn("route.table");
+
+ Map<String, Object> value =
+ ImmutableMap.of("route", ImmutableMap.of("table",
TABLE_IDENTIFIER.toString()));
+
+ List<IcebergWriterResult> writerResults = sinkWriterTest(value, config);
+ assertThat(writerResults).hasSize(1);
+
assertThat(writerResults.get(0).tableReference().identifier()).isEqualTo(TABLE_IDENTIFIER);
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testDynamicRouteReplaceNullWithDefault(boolean
replaceNullWithDefault) {
