apache/iceberg · error · UnsupportedOperationException

Schema not support for DMS records

Error message

Schema not support for DMS records

What it means

The DmsTransform SMT only supports schemaless Kafka Connect records: if record.value() is non-null and record.valueSchema() is non-null (a record WITH schema), it throws UnsupportedOperationException('Schema not support for DMS records'). AWS DMS records are expected as raw JSON map values without a Connect schema.

Solutions

1. Set value.converter.schemas.enable=false on the connector so values are schemaless maps
2. Switch the value converter to org.apache.kafka.connect.json.JsonConverter with schemas.enable=false
3. Remove DmsTransform and use a different transform if you need schema-bearing records
4. Preprocess records to strip value schemas before the transform

Example fix

// before (worker/connector config)
"value.converter.schemas.enable": "true"
// after
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false"

boolean dmsTransformCompatible(SinkRecord r) {
  return r.value() == null || r.valueSchema() == null;
}

boolean isSchemaless(SinkRecord r) {
  return r.valueSchema() == null && r.value() instanceof Map;
}

try {
  return dmsTransform.apply(record);
} catch (UnsupportedOperationException e) {
  throw new ConnectException("DmsTransform requires schemaless values; set value.converter.schemas.enable=false", e);
}

When it happens

Trigger: Configuring value.converter=JsonConverter with schemas.enable=true (or AvroConverter) upstream of DmsTransform, so records arrive with a non-null valueSchema.

Common situations: AWS DMS → Kafka Connect pipelines where the value converter emits schemas; switching converters from schemaless JSON to schema-bearing formats; Kafka Connect defaults where schemas.enable=true is set globally.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• Could not parse for
• Expected json object for record.value after parsing
• ${msg}
• Must specify table name(s)
• record value is not a string, use StringConverter

AI-assisted analysis of apache/iceberg@86d9c8fc54 (2026-09-12). Data as JSON: /api/errors/32123b4a88cf64bd. Report an issue: GitHub.

Appendix: source

Thrown at kafka-connect/kafka-connect-transforms/src/main/java/org/apache/iceberg/connect/transforms/DmsTransform.java:42

import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.transforms.util.Requirements;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DmsTransform<R extends ConnectRecord<R>> implements Transformation<R> {

  private static final Logger LOG = LoggerFactory.getLogger(DmsTransform.class.getName());
  private static final ConfigDef EMPTY_CONFIG = new ConfigDef();

  @Override
  public R apply(R record) {
    if (record.value() == null) {
      return record;
    } else if (record.valueSchema() == null) {
      return applySchemaless(record);
    } else {
      throw new UnsupportedOperationException("Schema not support for DMS records");
    }
  }

  @SuppressWarnings("unchecked")
  private R applySchemaless(R record) {
    Map<String, Object> value = Requirements.requireMap(record.value(), "DMS transform");

    // promote fields under "data"
    Object dataObj = value.get("data");
    Object metadataObj = value.get("metadata");
    if (!(dataObj instanceof Map) || !(metadataObj instanceof Map)) {
      LOG.debug("Unable to transform DMS record, skipping...");
      return null;
    }

    Map<String, Object> metadata = (Map<String, Object>) metadataObj;

    String dmsOp = metadata.get("operation").toString();

View on GitHub (pinned to 86d9c8fc54)