apache/beam · error · IllegalArgumentException

Unable to convert Kafka field schema

Error message

Unable to convert Kafka field schema %s to Beam Schema

What it means

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.

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.

Defensive patterns

Strategy: type-guard

Validate before calling

for (Field f : connectSchema.fields()) {
  Schema.Type t = f.schema().type();
  if (t != Schema.Type.ARRAY && t != Schema.Type.MAP && !isSupportedPrimitive(t)) {
    throw new IllegalStateException("Unsupported Connect type: " + t);
  }
}

Type guard

boolean isSupportedType(Schema s) {
  switch (s.type()) {
    case ARRAY: case MAP: return isSupportedType(s.valueSchema());
    default: return isSupportedPrimitive(s.type());
  }
}

Try / catch

try {
  Schema beamSchema = KafkaConnectUtils.beamSchema(connectSchema);
} catch (IllegalArgumentException e) {
  if (e.getMessage().startsWith("Unable to convert Kafka field schema")) {
    // drop or transform the offending field and retry
  }
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/0671d307a4bf2ce7. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaConnectUtils.java:76

        return Schema.FieldType.INT64;
      case FLOAT32:
        return Schema.FieldType.FLOAT;
      case FLOAT64:
        return Schema.FieldType.DOUBLE;
      case BOOLEAN:
        return Schema.FieldType.BOOLEAN;
      case STRING:
        return Schema.FieldType.STRING;
      case BYTES:
        return Schema.FieldType.BYTES;
      case ARRAY:
        return Schema.FieldType.array(beamSchemaTypeFromKafkaType(kafkaFieldSchema.valueSchema()));
      case MAP:
        return Schema.FieldType.map(
            beamSchemaTypeFromKafkaType(kafkaFieldSchema.keySchema()),
            beamSchemaTypeFromKafkaType(kafkaFieldSchema.valueSchema()));
      default:
        throw new IllegalArgumentException(
            String.format(
                "Unable to convert Kafka field schema %s to Beam Schema", kafkaFieldSchema));
    }
  }

  public static Instant debeziumRecordInstant(SourceRecord record) {
    if (record.valueSchema() != null
        && record.valueSchema().type().equals(org.apache.kafka.connect.data.Schema.Type.STRUCT)
        && record.valueSchema().field("ts_ms") != null
        && record.value() != null) {
      Struct recordValue = (Struct) record.value();
      return Instant.ofEpochMilli(recordValue.getInt64("ts_ms"));
    }

    if (record.sourceOffset() != null && record.sourceOffset().containsKey("ts_usec")) {
      Object tsUsecValue = record.sourceOffset().get("ts_usec");
      if (tsUsecValue instanceof Number) {
        return Instant.ofEpochMilli(((Number) tsUsecValue).longValue() / 1000);

View on GitHub (pinned to 12126d8942)