Module: ActiveRecord::Materialized::DebeziumEnvelope

Defined in:
lib/activerecord/materialized/debezium_envelope.rb

Overview

Maps a Debezium change envelope to an ingest_change descriptor. Pure and dependency-free: no Kafka / Debezium / Kafka-Connect runtime — just a mapping over the decoded envelope hash a consumer already holds. It removes the boilerplate (and the easy-to-miss cases: snapshot reads, tombstones, the nested/unwrapped shapes) every consumer otherwise hand-writes.

The envelope shape (a Debezium MySQL/Postgres connector event, whether or not the ExtractNewRecordState "unwrap" SMT has been applied — a nested payload is unwrapped here):

{ "op" => "u",
"before" => { "id" => 1, "category" => "books", ... },   # full row image (see below)
"after"  => { "id" => 1, "category" => "games", ... },
"ts_ms"  => 1710000000000,                               # connector processing time (unused)
"source" => { "table" => "line_items", "ts_ms" => 1710000000000, ... } }

The source.ts_ms (the DB commit time, monotonic within a source) is forwarded as the source_ts watermark, so a consumer relaying envelopes gets out-of-order suppression + freshness (SourceWatermark) for free. The top-level ts_ms is a different clock (the connector's processing time), so it is not used as a fallback — mixing clocks would break monotonicity. Correct partition-moving updates require a full before image (+binlog-row-image=FULL+ on MySQL, REPLICA IDENTITY FULL on Postgres); with only the primary key in before, the old partition is under-maintained until reconciliation heals it. See docs/out-of-band-writes.md and the CDC section of the README. This adapter is Debezium-specific; a Maxwell/other envelope (no op) raises rather than being silently dropped.

Class Method Summary collapse

Class Method Details

.to_change_descriptor(envelope, table = nil) ⇒ Hash?

Returns { table:, operation:, before:, after: } (plus source_ts: when the envelope carries a usable ts_ms) for ActiveRecord::Materialized.ingest_change, or nil for a tombstone (nothing to relay).

Parameters:

  • envelope (Hash, nil) —

    a decoded Debezium change event (string or symbol keys), or nil for a Kafka tombstone (the null-value message emitted after a delete for log compaction)

  • table (String, Symbol, nil) (defaults to: nil) —

    target-table override; defaults to the envelope's source.table

Returns:

  • (Hash, nil) —

    { table:, operation:, before:, after: } (plus source_ts: when the envelope carries a usable ts_ms) for ActiveRecord::Materialized.ingest_change, or nil for a tombstone (nothing to relay)

Raises:

  • (ArgumentError) —

    for a non-tombstone envelope with no op (not a Debezium change event, or not unwrapped), an unsupported op, or when the target table can't be determined



41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/activerecord/materialized/debezium_envelope.rb', line 41

def self.to_change_descriptor(envelope, table = nil)
  return nil if envelope.nil? # Kafka tombstone

  change = unwrap(envelope)
  op = fetch(change, "op")
  raise ArgumentError, "not a Debezium change envelope (no op)" if op.nil?

  descriptor = { table: resolve_table(table, change), operation: operation_for(op),
                 before: fetch(change, "before"), after: fetch(change, "after") }
  ts = source_ts(change)
  ts ? descriptor.merge(source_ts: ts) : descriptor
end