kafka-connector | org.apache.kafka.connect.storage.StringConverter for value not working

[raphaelauv]

Apache Iceberg version

1.7.1 (latest release)

Query engine

Kafka Connect

Please describe the bug 🐞

if I produce a message in string for key and value

then the connector

{
  "name": "string-event",
  "config": {
    "connector.class": "org.apache.iceberg.connect.IcebergSinkConnector",
    "tasks.max": "1",
    "topics": "aaaaaaa",
    "iceberg.tables.auto-create-enabled": "true",
    "iceberg.tables": "default.events",
    "iceberg.catalog.type": "rest",
    "iceberg.catalog.uri": "http://iceberg_rest:8181",
    "iceberg.catalog.warehouse": "s3://test-bucket/iceberg/",
    "iceberg.catalog.s3.endpoint": "http://minio:9000",
    "iceberg.catalog.s3.path-style-access": "true",
    "iceberg.catalog.s3.access-key-id": "${file:/secrets/credentials.properties:AWS_ACCESS_KEY_ID}",
    "iceberg.catalog.s3.secret-access-key": "${file:/secrets/credentials.properties:AWS_SECRET_ACCESS_KEY}",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "iceberg.control.commit.interval-ms": "10000"
  }
}

fail with

Caused by: java.lang.IllegalArgumentException: Not a struct type: string
at org.apache.iceberg.types.Type.asStructType(Type.java:72)
at org.apache.iceberg.connect.data.IcebergWriterFactory.autoCreateTable(IcebergWriterFactory.java:83)
at org.apache.iceberg.connect.data.IcebergWriterFactory.createWriter(IcebergWriterFactory.java:62)
at org.apache.iceberg.connect.data.SinkWriter.lambda$writerForTable$3(SinkWriter.java:139)
at java.base/java.util.HashMap.computeIfAbsent(Unknown Source)
at org.apache.iceberg.connect.data.SinkWriter.writerForTable(SinkWriter.java:138)
at org.apache.iceberg.connect.data.SinkWriter.lambda$routeRecordStatically$1(SinkWriter.java:98)
at java.base/java.util.Arrays$ArrayList.forEach(Unknown Source)
at org.apache.iceberg.connect.data.SinkWriter.routeRecordStatically(SinkWriter.java:96)
at org.apache.iceberg.connect.data.SinkWriter.save(SinkWriter.java:85)
at java.base/java.util.ArrayList.forEach(Unknown Source)
at org.apache.iceberg.connect.data.SinkWriter.save(SinkWriter.java:68)
at org.apache.iceberg.connect.channel.Worker.save(Worker.java:124)
at org.apache.iceberg.connect.channel.CommitterImpl.save(CommitterImpl.java:88)
at org.apache.iceberg.connect.IcebergSinkTask.put(IcebergSinkTask.java:87)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:605)

even with a string value containing a json it fail

but if I use

{
"value.converter":"org.apache.kafka.connect.json.JsonConverter"
"value.converter.schemas.enable":"true"
}

then the connector create and add data to the table ->

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time