laskoviymishka commented on code in PR #16598:
URL: https://github.com/apache/iceberg/pull/16598#discussion_r3441690182
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordUtils.java:
##########
@@ -122,7 +123,14 @@ public static TaskWriter<Record> createTableWriter(
colName -> {
NestedField field = table.schema().findField(colName);
if (field == null) {
- throw new IllegalArgumentException("ID column not found:
" + colName);
+ throw new DataException(
Review Comment:
This is the path I'd most want closed before merge. `createTableWriter` is
the common writer-creation point for both new and existing tables, and it never
checks `schemaForceOptional()`. So with `schema-force-optional=true` and a
per-table `id-columns` against a table that already exists, neither the startup
check (only guards the global `default-id-columns` key) nor the auto-create
check (only fires when the table is absent) runs — the connector silently
configures an equality-delete writer over fields that schema-force-optional
will later flip to optional.
I'd add the same `idColumns`-non-empty + `config.schemaForceOptional()`
guard right here, throwing the same DataException as the auto-create path. That
makes this the single enforcement point and closes the existing-table hole at
the same time. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java:
##########
@@ -282,6 +282,10 @@ private void validate() {
} else {
throw new ConfigException("Must specify table name(s)");
}
+ checkState(
Review Comment:
This only inspects `tablesDefaultIdColumns()`, the global key. Per-table
id-columns live under `iceberg.table.<name>.id-columns` and never appear here,
so `schema-force-optional=true` + `iceberg.table.db.orders.id-columns=id` (no
global default) passes startup clean.
If we move the real enforcement into `createTableWriter` (see the
RecordUtils comment), that covers both new and existing tables at runtime and
I'd be comfortable leaving this startup check as the fast-fail for the global
case. Otherwise this check needs to also scan `originalProps` for per-table
id-columns keys. Either is fine, but the two paths together shouldn't leave a
gap.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java:
##########
@@ -282,6 +282,10 @@ private void validate() {
} else {
throw new ConfigException("Must specify table name(s)");
}
+ checkState(
+ !(schemaForceOptional() && tablesDefaultIdColumns() != null),
Review Comment:
`tablesDefaultIdColumns()` is backed by `getString`, so
`iceberg.tables.default-id-columns=` (empty) returns `""`, which is `!= null` —
the ConfigException fires even though `stringToList("", ",")` is empty and no
id-columns are actually set.
I'd guard on the parsed list instead: `!(schemaForceOptional() &&
!stringToList(tablesDefaultIdColumns(), ",").isEmpty())`.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordUtils.java:
##########
@@ -19,29 +19,87 @@
package org.apache.iceberg.connect.data;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import java.util.Map;
+import java.util.UUID;
+import org.apache.iceberg.LocationProviders;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.connect.IcebergSinkConfig;
+import org.apache.iceberg.connect.TableSinkConfig;
+import org.apache.iceberg.connect.events.TableReference;
+import org.apache.iceberg.encryption.PlaintextEncryptionManager;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.kafka.connect.data.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.types.Types;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.errors.DataException;
import org.junit.jupiter.api.Test;
public class TestRecordUtils {
+ @Test
+ public void testCreateTableWriterMissingIdColumnThrowsDataException() {
Review Comment:
There's no test exercising `createTableWriter` with
`schemaForceOptional=true` and valid id-columns on an existing table — this
covers only the missing-column case. That's the same blind spot as the
validation gap in `createTableWriter`: whatever we decide there (throw a
DataException, or document the accepted behavior), I'd lock it with a test so
the existing-table contract doesn't regress silently.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestIcebergWriterFactory.java:
##########
@@ -96,4 +101,181 @@ public void testAutoCreateTable(boolean partitioned) {
assertThat(capturedArguments.get(1)).isEqualTo(Namespace.of("foo1",
"foo2"));
assertThat(capturedArguments.get(2)).isEqualTo(Namespace.of("foo1",
"foo2", "foo3"));
}
+
+ // schema-force-optional=true with id-columns configured: Kafka field is
required but the config
+ // flag forces every Iceberg field optional, so the combination is rejected
at the connector
+ // layer.
+ @Test
+ @SuppressWarnings("unchecked")
Review Comment:
The five new `@Test` methods all carry `@SuppressWarnings("unchecked")`, but
that annotation only exists because `testAutoCreateTable` uses
`ArgumentCaptor.forClass(Map.class)`. None of the new methods use a raw type or
unchecked cast, so the suppression was copied verbatim and now masks any
legitimate future unchecked warning in those methods.
I'd drop `@SuppressWarnings("unchecked")` from the five new tests and keep
it only on `testAutoCreateTable`.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/IcebergWriterFactory.java:
##########
@@ -93,7 +96,54 @@ Table autoCreateTable(String tableName, SinkRecord sample) {
structType = SchemaUtils.toIcebergType(sample.valueSchema(),
config).asStructType();
}
- org.apache.iceberg.Schema schema = new
org.apache.iceberg.Schema(structType.fields());
+ List<String> idColumns = config.tableConfig(tableName).idColumns();
+
+ if (!idColumns.isEmpty() && config.schemaForceOptional()) {
+ throw new DataException(
+ String.format(
+ "iceberg.tables.schema-force-optional is enabled for table %s
but id-columns are configured. "
+ + "schema-force-optional marks every field optional, which
is incompatible with identifier fields that must be required. "
+ + "Disable schema-force-optional or remove the id-columns
configuration.",
+ tableName));
+ }
+
+ Set<Integer> identifierFieldIds =
+ idColumns.stream()
+ .map(
+ name -> {
+ if (name.contains(".")) {
Review Comment:
The spec allows nested identifier fields — `format/spec.md`: "Identifier
fields may be nested in structs but cannot be nested within maps or lists" —
and the existing-table path already supports them, since
`RecordUtils.createTableWriter` resolves columns with
`table.schema().findField(colName)`, which handles dotted paths like `user.id`.
So `id-columns=user.id` succeeds on a pre-existing table and throws here on
auto-create. The real constraint is just that `structType.field(name)` is
top-level-only, which is an implementation detail, not a connector-wide
limitation.
I'd either split on "." and walk the nested `StructType` to the leaf field
id (matching `findField`), or, if we want to keep auto-create simple for now,
reword to "the connector's auto-create path currently supports only top-level
identifier fields; nested fields can be configured on pre-existing tables" and
drop "not supported by the connector." Right now the message claims a
restriction that isn't true. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordUtils.java:
##########
@@ -122,7 +123,14 @@ public static TaskWriter<Record> createTableWriter(
colName -> {
NestedField field = table.schema().findField(colName);
if (field == null) {
- throw new IllegalArgumentException("ID column not found:
" + colName);
+ throw new DataException(
+ String.format(
+ "ID column '%s' not found in schema for table
%s. Available columns: %s",
+ colName,
+ tableReference.identifier().name(),
Review Comment:
`TableIdentifier.name()` returns only the unqualified name, so this message
says "for table orders" while the identical error in `IcebergWriterFactory`
uses the full `tableName` ("db.orders"). For a production DataException the
namespace is exactly what an operator needs, and it's ambiguous when the same
unqualified name exists across namespaces.
I'd pass `tableReference.identifier()` so `toString()` includes the
namespace.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/IcebergWriterFactory.java:
##########
@@ -93,7 +96,54 @@ Table autoCreateTable(String tableName, SinkRecord sample) {
structType = SchemaUtils.toIcebergType(sample.valueSchema(),
config).asStructType();
}
- org.apache.iceberg.Schema schema = new
org.apache.iceberg.Schema(structType.fields());
+ List<String> idColumns = config.tableConfig(tableName).idColumns();
+
+ if (!idColumns.isEmpty() && config.schemaForceOptional()) {
+ throw new DataException(
+ String.format(
+ "iceberg.tables.schema-force-optional is enabled for table %s
but id-columns are configured. "
+ + "schema-force-optional marks every field optional, which
is incompatible with identifier fields that must be required. "
+ + "Disable schema-force-optional or remove the id-columns
configuration.",
+ tableName));
+ }
+
+ Set<Integer> identifierFieldIds =
+ idColumns.stream()
+ .map(
Review Comment:
The `.map()` lambda throws on two branches before returning a field id.
`map` is meant to be non-interfering and stateless, so the exception surfaces
from inside `collect()` and the stack trace points into `Collectors.toSet`, not
a named method — and Errorprone/SpotBugs flag throwing from stream lambdas.
I'd make this an explicit `for` loop over `idColumns` building the
`Set<Integer>`. It reads more directly given the two validation branches and
gives a clean stack frame when it throws.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
