{"record":{"id":"8f81f1b4393d2159","repo":"apache/iceberg","slug":"cannot-convert-type-data-getclass-getname","errorCode":null,"errorMessage":"Cannot convert type: ${data.getClass().getName()}","messagePattern":"Cannot convert type: (.+?)","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java","lineNumber":118,"sourceCode":"  private final NameMapping nameMapping;\n  private final IcebergSinkConfig config;\n  private final Map<Integer, Map<String, NestedField>> structNameMap = Maps.newHashMap();\n\n  RecordConverter(Table table, IcebergSinkConfig config) {\n    this.tableSchema = table.schema();\n    this.nameMapping = createNameMapping(table);\n    this.config = config;\n  }\n\n  Record convert(Object data) {\n    return convert(data, null);\n  }\n\n  Record convert(Object data, SchemaUpdate.Consumer schemaUpdateConsumer) {\n    if (data instanceof Struct || data instanceof Map) {\n      return convertStructValue(data, tableSchema.asStruct(), -1, schemaUpdateConsumer);\n    }\n    throw new UnsupportedOperationException(\"Cannot convert type: \" + data.getClass().getName());\n  }\n\n  private NameMapping createNameMapping(Table table) {\n    String nameMappingString = table.properties().get(TableProperties.DEFAULT_NAME_MAPPING);\n    return nameMappingString != null ? NameMappingParser.fromJson(nameMappingString) : null;\n  }\n\n  private Object convertValue(\n      Object value, Type type, int fieldId, SchemaUpdate.Consumer schemaUpdateConsumer) {\n    if (value == null) {\n      return null;\n    }\n    switch (type.typeId()) {\n      case STRUCT:\n        return convertStructValue(value, type.asStructType(), fieldId, schemaUpdateConsumer);\n      case LIST:\n        return convertListValue(value, type.asListType(), schemaUpdateConsumer);\n      case MAP:","sourceCodeStart":100,"sourceCodeEnd":136,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java#L100-L136","documentation":"RecordConverter.convert(Object, SchemaUpdate.Consumer) only accepts Kafka Connect Struct or Map values representing the top-level record. Any other Java type (e.g. primitives, lists, raw strings) cannot represent an Iceberg table row, so it throws UnsupportedOperationException naming the actual value class.","triggerScenarios":"Calling RecordConverter.convert with a value that is neither org.apache.kafka.connect.data.Struct nor java.util.Map, e.g. convert(\"some string\", consumer), convert(List.of(1,2), consumer), or passing a numeric/byte[] payload.","commonSituations":"Kafka Connect record values that were not wrapped in a struct — plain JSON scalars/arrays from topics ingested via the Iceberg sink, or custom users of the connector API passing raw values instead of structured records.","solutions":["Ensure the Kafka record value is deserialized into a Connect Struct (set an appropriate value converter, e.g. JsonSchemaConverter with schema, or a schema-registry-backed converter).","Wrap the raw value in a Map (map of field name -> value) so it goes down the convertStructValue path.","Catch UnsupportedOperationException at the record level and route the record to a DLQ instead of failing the task."],"exampleFix":"// before\nRecord r = converter.convert(record.value(), consumer); // value is a String\n// after\nMap<String, Object> wrapped = Collections.singletonMap(\"payload\", record.value());\nRecord r = converter.convert(wrapped, consumer);","handlingStrategy":"type-guard","validationCode":"if (!(value instanceof Struct) && !(value instanceof Map)) {\n  throw new IllegalArgumentException(\"Top-level record must be Struct or Map, got \" + value.getClass().getName());\n}","typeGuard":"boolean isConvertible(Object v) {\n  return v instanceof Struct || v instanceof Map;\n}","tryCatchPattern":"try {\n  Record r = converter.convert(value, consumer);\n} catch (UnsupportedOperationException e) {\n  log.error(\"Unsupported record value type\", e);\n  sendToDlq(record, e);\n}","preventionTips":["Use a schema-registry-backed value converter so Connect records arrive as Structs.","Verify the topic's Connect value schema type (struct) in connector config before running.","Log record.value().getClass() in a test SMT to confirm payload shape before deploying."],"tags":["kafka-connect","unsupported-operation","type-mismatch"],"backgroundTag":"incompatible-source-type","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}