apache/beam · error · IllegalArgumentException

Debezium record received is not of the right kind. Should…

Error message

Debezium record received is not of the right kind. Should be STRUCT with ts_ms field or sourceOffset with ts_usec. Instead it is: %s, %s

What it means

debeziumRecordInstant extracts an event timestamp from a Debezium SourceRecord. It expects either a STRUCT value with a ts_ms field or a sourceOffset containing ts_usec. If neither yields a usable timestamp it throws, reporting the record's valueSchema and sourceOffset for diagnosis.

Solutions

1. Disable tombstones: set 'tombstones.on.delete' to false in the Debezium connector config.
2. Filter out records with null value or non-STRUCT schema before calling debeziumRecordInstant (e.g. via a DoFn guard).
3. Upgrade Beam's KafkaConnectUtils, which handles more record kinds.
4. If timestamps matter, enable Debezium heartbeats so ts_ms is populated consistently.

Example fix

// before
Instant ts = KafkaConnectUtils.debeziumRecordInstant(record);
// after
if (record.value() instanceof Struct) {
  Instant ts = KafkaConnectUtils.debeziumRecordInstant(record);
} // else skip tombstone/heartbeat record

boolean hasTimestamp(SourceRecord r) {
  if (r.value() instanceof Struct) {
    return ((Struct) r.value()).schema().field("ts_ms") != null;
  }
  Object ts = r.sourceOffset() != null ? r.sourceOffset().get("ts_usec") : null;
  return ts instanceof Number;
}

boolean isTimestampedStruct(SourceRecord r) {
  return r.value() instanceof Struct
      && ((Struct) r.value()).schema().field("ts_ms") != null;
}

try {
  Instant ts = KafkaConnectUtils.debeziumRecordInstant(record);
} catch (IllegalArgumentException e) {
  if (e.getMessage().startsWith("Debezium record received is not of the right kind")) {
    return; // skip tombstone/heartbeat record
  }
  throw e;
}

When it happens

Trigger: Polling a Debezium record whose value is not a STRUCT with ts_ms (e.g. a tombstone record with null value/schema, or a heartbeat/schema-change record) and whose sourceOffset lacks a numeric ts_usec entry.

Common situations: Enabling tombstone events (tombstones.on.delete=true, the default), heartbeat records fromDebezium heartbeats, schema-change topics routed into the same pipeline, or connectors that omit ts_usec/ts_ms.

Related errors

• BigQuery data contained value
• Cannot create enum from value!
• could not fetch database schema
• Dropping output field
• Empty timestamp.

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

Appendix: source

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

  }

  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);
      }
    }

    throw new IllegalArgumentException(
        "Debezium record received is not of the right kind. "
            + String.format(
                "Should be STRUCT with ts_ms field or sourceOffset with ts_usec. Instead it is: %s, %s",
                record.valueSchema(), record.sourceOffset()));
  }

  public static SourceRecordMapper<Row> beamRowFromSourceRecordFn(final Schema recordSchema) {
    return new SourceRecordMapper<Row>() {
      @Override
      public Row mapSourceRecord(SourceRecord sourceRecord) throws Exception {
        return beamRowFromKafkaStruct((Struct) sourceRecord.value(), recordSchema);
      }

      private Row beamRowFromKafkaStruct(Struct kafkaStruct, Schema beamSchema) {
        Row.Builder rowBuilder = Row.withSchema(beamSchema);
        for (Schema.Field f : beamSchema.getFields()) {
          Object structField = kafkaStruct.getWithoutDefault(f.getName());
          switch (kafkaStruct.schema().field(f.getName()).schema().type()) {

View on GitHub (pinned to 12126d8942)