{"record":{"id":"69a47e7a6683ffb5","repo":"apache/beam","slug":"debezium-record-received-is-not-of-the-right-kind-should-be","errorCode":null,"errorMessage":"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","messagePattern":"Debezium record received is not of the right kind\\. Should be STRUCT with ts_ms field or sourceOffset with ts_usec\\. Instead it is: (.+?), (.+?)","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaConnectUtils.java","lineNumber":98,"sourceCode":"  }\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);\n      }\n    }\n\n    throw new IllegalArgumentException(\n        \"Debezium record received is not of the right kind. \"\n            + String.format(\n                \"Should be STRUCT with ts_ms field or sourceOffset with ts_usec. Instead it is: %s, %s\",\n                record.valueSchema(), record.sourceOffset()));\n  }\n\n  public static SourceRecordMapper<Row> beamRowFromSourceRecordFn(final Schema recordSchema) {\n    return new SourceRecordMapper<Row>() {\n      @Override\n      public Row mapSourceRecord(SourceRecord sourceRecord) throws Exception {\n        return beamRowFromKafkaStruct((Struct) sourceRecord.value(), recordSchema);\n      }\n\n      private Row beamRowFromKafkaStruct(Struct kafkaStruct, Schema beamSchema) {\n        Row.Builder rowBuilder = Row.withSchema(beamSchema);\n        for (Schema.Field f : beamSchema.getFields()) {\n          Object structField = kafkaStruct.getWithoutDefault(f.getName());\n          switch (kafkaStruct.schema().field(f.getName()).schema().type()) {","sourceCodeStart":80,"sourceCodeEnd":116,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaConnectUtils.java#L80-L116","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Disable tombstones: set 'tombstones.on.delete' to false in the Debezium connector config.","Filter out records with null value or non-STRUCT schema before calling debeziumRecordInstant (e.g. via a DoFn guard).","Upgrade Beam's KafkaConnectUtils, which handles more record kinds.","If timestamps matter, enable Debezium heartbeats so ts_ms is populated consistently."],"exampleFix":"// before\nInstant ts = KafkaConnectUtils.debeziumRecordInstant(record);\n// after\nif (record.value() instanceof Struct) {\n  Instant ts = KafkaConnectUtils.debeziumRecordInstant(record);\n} // else skip tombstone/heartbeat record","handlingStrategy":"type-guard","validationCode":"boolean hasTimestamp(SourceRecord r) {\n  if (r.value() instanceof Struct) {\n    return ((Struct) r.value()).schema().field(\"ts_ms\") != null;\n  }\n  Object ts = r.sourceOffset() != null ? r.sourceOffset().get(\"ts_usec\") : null;\n  return ts instanceof Number;\n}","typeGuard":"boolean isTimestampedStruct(SourceRecord r) {\n  return r.value() instanceof Struct\n      && ((Struct) r.value()).schema().field(\"ts_ms\") != null;\n}","tryCatchPattern":"try {\n  Instant ts = KafkaConnectUtils.debeziumRecordInstant(record);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().startsWith(\"Debezium record received is not of the right kind\")) {\n    return; // skip tombstone/heartbeat record\n  }\n  throw e;\n}","preventionTips":["Set tombstones.on.delete=false in the Debezium connector config","Filter records with null value before timestamp extraction","Enable heartbeats so ts_ms is present on quiet topics","Handle schema-change records on a separate topic"],"tags":["java","debezium","kafka-connect","timestamp"],"backgroundTag":"unexpected-response-shape","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"}