Skip to content

add dynamic routing with namespace awareness #14553

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

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open

add dynamic routing with namespace awareness #14553

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
668d681
add dynamic routing with namespace awareness
djakupovic Nov 10, 2025
3375481
fix test assert
djakupovic Nov 10, 2025
b06cff0
fix formatting
djakupovic Nov 10, 2025
6b4de88
added route prefix for dynamic routing
djakupovic Nov 10, 2025
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
22 changes: 22 additions & 0 deletions kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,8 @@ public class IcebergSinkConfig extends AbstractConfig {
private static final String CATALOG_NAME_PROP = "iceberg.catalog";
private static final String TABLES_PROP = "iceberg.tables";
private static final String TABLES_DYNAMIC_PROP = "iceberg.tables.dynamic-enabled";
private static final String TABLES_ROUTE_NAMESPACE_PROP = "iceberg.tables.route-namespace";
private static final String TABLES_ROUTE_PREFIX_PROP = "iceberg.tables.route-prefix";
private static final String TABLES_ROUTE_FIELD_PROP = "iceberg.tables.route-field";
private static final String TABLES_DEFAULT_COMMIT_BRANCH = "iceberg.tables.default-commit-branch";
private static final String TABLES_DEFAULT_ID_COLUMNS = "iceberg.tables.default-id-columns";
Expand Down Expand Up @@ -127,6 +129,18 @@ private static ConfigDef newConfigDef() {
false,
Importance.MEDIUM,
"Enable dynamic routing to tables based on a record value");
configDef.define(
TABLES_ROUTE_NAMESPACE_PROP,
ConfigDef.Type.STRING,
null,
Importance.MEDIUM,
"Target namespace for routing records to tables");
configDef.define(
TABLES_ROUTE_PREFIX_PROP,
ConfigDef.Type.STRING,
null,
Importance.MEDIUM,
"Target prefix for routing records to tables");
configDef.define(
TABLES_ROUTE_FIELD_PROP,
ConfigDef.Type.STRING,
Expand Down Expand Up @@ -335,6 +349,14 @@ public boolean dynamicTablesEnabled() {
return getBoolean(TABLES_DYNAMIC_PROP);
}

public String tablesRouteNamespace() {
return getString(TABLES_ROUTE_NAMESPACE_PROP);
}

public String tablesRoutePrefix() {
return getString(TABLES_ROUTE_PREFIX_PROP);
}

public String tablesRouteField() {
return getString(TABLES_ROUTE_FIELD_PROP);
}
Expand Down
10 changes: 9 additions & 1 deletion kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,12 +115,20 @@ private void routeRecordStatically(SinkRecord record) {
}

private void routeRecordDynamically(SinkRecord record) {
String routeNamespace = config.tablesRouteNamespace();
String routeField = config.tablesRouteField();
Preconditions.checkNotNull(routeField, "Route field cannot be null with dynamic routing");

String routeValue = extractRouteValue(record.value(), routeField);
if (config.tablesRoutePrefix() != null) {
routeValue = config.tablesRoutePrefix() + "_" + routeValue;
}
if (routeValue != null) {
String tableName = routeValue.toLowerCase(Locale.ROOT);
String tableName =
(routeNamespace != null)
? routeNamespace.toLowerCase() + "." + routeValue.toLowerCase(Locale.ROOT)
: routeValue.toLowerCase(Locale.ROOT);

writerForTable(tableName, record, true).write(record);
}
}
Expand Down
18 changes: 18 additions & 0 deletions ...a-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ public class TestSinkWriter {
optional(1, "id", Types.LongType.get()),
optional(2, "data", Types.StringType.get()),
optional(3, "date", Types.StringType.get()));
private static final String ROUTE_NAMESPACE = "namespace";
private static final String ROUTE_FIELD = "fld";

@BeforeEach
Expand Down Expand Up @@ -153,6 +154,23 @@ public void testDynamicRoute() {
assertThat(writerResult.tableIdentifier()).isEqualTo(TABLE_IDENTIFIER);
}

@Test
public void testDynamicNamespaceRoute() {
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.tablesRouteNamespace()).thenReturn(ROUTE_NAMESPACE);
when(config.tablesRouteField()).thenReturn(ROUTE_FIELD);

Map<String, Object> value = ImmutableMap.of(ROUTE_FIELD, ROUTE_NAMESPACE);

List<IcebergWriterResult> writerResults = sinkWriterTest(value, config);
assertThat(writerResults).hasSize(1);
IcebergWriterResult writerResult = writerResults.get(0);
assertThat(writerResult.tableIdentifier()).isEqualTo(ROUTE_NAMESPACE + "." + ROUTE_FIELD);
}

@Test
public void testDynamicNoRoute() {
IcebergSinkConfig config = mock(IcebergSinkConfig.class);
Expand Down
Loading