{"record":{"id":"d6ced4ca524ef75b","repo":"apache/iceberg","slug":"record-schema-of-type-does-not-match-table-of-t","errorCode":null,"errorMessage":"Record schema of type {} does not match table of type {}","messagePattern":"Record schema of type (.+?) does not match table of type (.+?)","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java","lineNumber":361,"sourceCode":"        if (tableType.isMapType()) {\n          MapType mapType = tableType.asMapType();\n          evolveSchemaFromConnectSchema(\n              recordSchema.valueSchema(),\n              mapType.valueType(),\n              mapType.valueId(),\n              schemaUpdateConsumer);\n        } else {\n          logMismatchedType(recordSchema.type(), tableType);\n        }\n        break;\n      default:\n        break;\n    }\n  }\n\n  private void logMismatchedType(\n      org.apache.kafka.connect.data.Schema.Type recordSchemaType, Type tableType) {\n    LOG.warn(\n        \"Record schema of type {} does not match table of type {}\", recordSchemaType, tableType);\n  }\n\n  private NestedField lookupStructField(String fieldName, StructType schema, int structFieldId) {\n    if (nameMapping == null) {\n      return config.schemaCaseInsensitive()\n          ? schema.caseInsensitiveField(fieldName)\n          : schema.field(fieldName);\n    }\n\n    return structNameMap\n        .computeIfAbsent(structFieldId, notUsed -> createStructNameMap(schema))\n        .get(fieldName);\n  }\n\n  private Map<String, NestedField> createStructNameMap(StructType schema) {\n    Map<String, NestedField> map = Maps.newHashMap();\n    schema","sourceCodeStart":343,"sourceCodeEnd":379,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java#L343-L379","documentation":"RecordConverter.evolveSchemaFromConnectSchema logs a warning when the Kafka Connect record's field type does not match the corresponding Iceberg table column type, so automatic schema evolution cannot be applied for that field. The field is skipped rather than evolved, and data may be written with the existing table type or fail downstream conversion.","triggerScenarios":"Writing records via IcebergWriter with schema evolution (evolve-schema enabled) where a Connect field type (e.g. int, timestamp) does not correspond to the declared Iceberg type (e.g. table has long or string where record has int32).","commonSituations":"Producer changed a field's type (int -> long, string -> timestamp); table schema was manually altered; Connect schema inference derives a different logical type than the table's.","solutions":["Alter the Iceberg table schema so the column type matches the record schema","Adjust the upstream producer to emit the type the table expects","Use a schema transformation/message converter in Connect to coerce the record type","Check handling of logical types (timestamp/decimal) in the Connect schema — set appropriate config (e.g. timestamp conversion)"],"exampleFix":"// before: table column `count` is long, record schema type is int32 -> warning\n// after: evolve table to match\n// Spark: ALTER TABLE catalog.db.t ALTER COLUMN count TYPE long; or set count to int in the producer","handlingStrategy":"validation","validationCode":"org.apache.kafka.connect.data.Schema.Type connectType = recordSchema.field(name).schema().type();\nType icebergType = table.schema().findField(name).type();\nif (!RecordConverter.matches(connectType, icebergType)) { /* fix table or producer schema before writing */ }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Keep producer schema and table schema aligned via schema registry checks","Enable evolve-schema and review conversion warnings early in staging","Avoid manual table ALTERs that diverge from the producer contract"],"tags":["kafka-connect","schema-evolution","type-mismatch"],"backgroundTag":"type-mismatch","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"}