{"record":{"id":"0671d307a4bf2ce7","repo":"apache/beam","slug":"unable-to-convert-kafka-field-schema-s-to-beam-schema","errorCode":null,"errorMessage":"Unable to convert Kafka field schema %s to Beam Schema","messagePattern":"Unable to convert Kafka field schema (.+?) to Beam Schema","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaConnectUtils.java","lineNumber":76,"sourceCode":"        return Schema.FieldType.INT64;\n      case FLOAT32:\n        return Schema.FieldType.FLOAT;\n      case FLOAT64:\n        return Schema.FieldType.DOUBLE;\n      case BOOLEAN:\n        return Schema.FieldType.BOOLEAN;\n      case STRING:\n        return Schema.FieldType.STRING;\n      case BYTES:\n        return Schema.FieldType.BYTES;\n      case ARRAY:\n        return Schema.FieldType.array(beamSchemaTypeFromKafkaType(kafkaFieldSchema.valueSchema()));\n      case MAP:\n        return Schema.FieldType.map(\n            beamSchemaTypeFromKafkaType(kafkaFieldSchema.keySchema()),\n            beamSchemaTypeFromKafkaType(kafkaFieldSchema.valueSchema()));\n      default:\n        throw new IllegalArgumentException(\n            String.format(\n                \"Unable to convert Kafka field schema %s to Beam Schema\", kafkaFieldSchema));\n    }\n  }\n\n  public static Instant debeziumRecordInstant(SourceRecord record) {\n    if (record.valueSchema() != null\n        && record.valueSchema().type().equals(org.apache.kafka.connect.data.Schema.Type.STRUCT)\n        && record.valueSchema().field(\"ts_ms\") != null\n        && record.value() != null) {\n      Struct recordValue = (Struct) record.value();\n      return Instant.ofEpochMilli(recordValue.getInt64(\"ts_ms\"));\n    }\n\n    if (record.sourceOffset() != null && record.sourceOffset().containsKey(\"ts_usec\")) {\n      Object tsUsecValue = record.sourceOffset().get(\"ts_usec\");\n      if (tsUsecValue instanceof Number) {\n        return Instant.ofEpochMilli(((Number) tsUsecValue).longValue() / 1000);","sourceCodeStart":58,"sourceCodeEnd":94,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaConnectUtils.java#L58-L94","documentation":"beamSchemaTypeFromKafkaType maps Kafka Connect schema types to Beam Schema field types. ARRAY and MAP are handled recursively; any other unrecognized type (e.g. an unexpected composite type) falls into the default branch and throws. It indicates the Kafka Connect schema contains a type this converter does not support.","triggerScenarios":"Calling beamField/beamSchemaTypeFromKafkaType on a Connect Schema whose type() is not one of the handled primitives, ARRAY, or MAP — typically an exotic or schema-less (by-name referenced) type in the Debezium connector's emitted schema.","commonSituations":"Using a Debezium connector or single-message transforms that emit unusual field types; upgrading Debezium so new types appear; custom converters emitting non-standard schemas.","solutions":["Identify the offending field from the message and exclude/flatten it before conversion.","Upgrade Apache Beam, which may add support for more Connect types.","Pre-transform the SourceRecord schema to standard types before calling KafkaConnectUtils.","File/patch KafkaConnectUtils to map the missing type."],"exampleFix":null,"handlingStrategy":"type-guard","validationCode":"for (Field f : connectSchema.fields()) {\n  Schema.Type t = f.schema().type();\n  if (t != Schema.Type.ARRAY && t != Schema.Type.MAP && !isSupportedPrimitive(t)) {\n    throw new IllegalStateException(\"Unsupported Connect type: \" + t);\n  }\n}","typeGuard":"boolean isSupportedType(Schema s) {\n  switch (s.type()) {\n    case ARRAY: case MAP: return isSupportedType(s.valueSchema());\n    default: return isSupportedPrimitive(s.type());\n  }\n}","tryCatchPattern":"try {\n  Schema beamSchema = KafkaConnectUtils.beamSchema(connectSchema);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().startsWith(\"Unable to convert Kafka field schema\")) {\n    // drop or transform the offending field and retry\n  }\n}","preventionTips":["Keep Beam upgraded so new Connect types are supported","Restrict ingested columns to supported types via connector exclude lists","Test schema conversion against your table's schema before production"],"tags":["java","schema","kafka-connect","apache-beam"],"backgroundTag":"incompatible-source-type","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}