dbt developer on retainer: stg_shipment_events incremental model max(event_timestamp) filter permanently excludes 847,293 Kafka backlog events — fct_shipments missing 3 days of delivery confirmations for 45 days

dbt Core 1.7.4 • Snowflake Enterprise Edition • Kafka Connect • incremental materialization • max(event_timestamp) filter • late-arriving records • _fivetran_synced • full-refresh backfill • retainer billing

Meridian Freight Solutions had been receiving accurate carrier SLA reports from its analytics platform for over a year. In October 2026, an accounts payable analyst running a carrier invoice reconciliation discovered that 2,847 shipments marked as “delivery unconfirmed” in the analytics system had delivery receipts in the carrier’s own portal. When the discrepancy was traced to the dbt pipeline, the retainer developer found that 847,293 shipment events from the last weekend of August 2026 were entirely absent from the stg_shipment_events model — and from every downstream mart that depended on it. The events existed in the Snowflake raw layer. Every dbt job run since their arrival had completed with zero errors and zero test failures. The events had been invisible in the analytics system for 45 days because the incremental model’s filter used the domain event timestamp rather than the ingestion timestamp, and a three-day Kafka broker migration had delivered the August 29–31 backlog with timestamps that the filter had permanently locked out.

Meridian Freight Solutions ran dbt Core 1.7.4 orchestrated by Apache Airflow 2.9.1 on Amazon MWAA. The Snowflake data warehouse ran in AWS us-east-1. Raw shipment event data arrived via a Kafka Connect connector that consumed from a Kafka cluster on a dedicated EC2 cluster and wrote to a Snowflake Kafka Connector sink. The Snowflake sink table, RAW.EVENTS.SHIPMENT_EVENTS, received one row per Kafka message. Each row included event_id (UUID), event_timestamp (the timestamp recorded by the source system when the event occurred), event_type (DEPARTED, IN_TRANSIT, DELIVERED, EXCEPTION), shipment_id, carrier_id, and _fivetran_synced (the timestamp when the Snowflake Kafka Connector wrote the row to the Snowflake table). In normal operation, _fivetran_synced was within 30–60 seconds of event_timestamp.

The dbt staging model stg_shipment_events was configured as an incremental model with unique_key: ‘event_id’ and strategy: ‘merge’. Its incremental filter used the canonical max-timestamp pattern:

-- models/staging/stg_shipment_events.sql (dbt Core 1.7.4)
-- Incremental model: strategy=merge, unique_key=event_id

{{
    config(
        materialized='incremental',
        unique_key='event_id',
        incremental_strategy='merge',
        on_schema_change='sync_all_columns'
    )
}}

SELECT
    event_id,
    event_timestamp,
    event_type,
    shipment_id,
    carrier_id,
    _fivetran_synced,
    CURRENT_TIMESTAMP() AS dbt_loaded_at
FROM {{ source('raw_events', 'shipment_events') }}

{% if is_incremental() %}
    WHERE event_timestamp > (
        SELECT MAX(event_timestamp)
        FROM {{ this }}
    )
{% endif %}

-- This filter was correct when:
--   _fivetran_synced ≈ event_timestamp (ingestion lag < 60 seconds)
-- This filter FAILS when:
--   events arrive late (Kafka backlog) with past event_timestamps
--   by the time they arrive, MAX(event_timestamp) has already advanced
--   past their event_timestamp → they are permanently excluded

On August 28, 2026, the platform engineering team began migrating the Kafka broker cluster from a three-node EC2 cluster to a new five-node cluster for expanded capacity. The migration required a rolling broker restart and a consumer group rebalance. During the migration window (August 29–31, the weekend), the Kafka Connect connector’s consumer group was in a rebalance loop and did not commit consumer offsets to the new broker cluster. Kafka events generated by the source system (warehouse management system, carrier APIs, and IoT GPS trackers) during August 29–31 were published to Kafka topics and retained on the broker but were not consumed by the Kafka Connect connector. These 847,293 events accumulated with their original event_timestamp values in the August 29–31 range.

How the incremental filter permanently locked out the backlog events

On September 1, 2026, the Kafka broker migration completed and the Kafka Connect connector successfully rebalanced and began consuming messages. The 847,293 backlogged events were consumed from the Kafka topics and delivered to RAW.EVENTS.SHIPMENT_EVENTS over the course of September 1 and 2. These rows had event_timestamp values of August 29–31 and _fivetran_synced values of September 1–2 (reflecting when they actually arrived in Snowflake).

The dbt Airflow DAG ran daily at 02:00 UTC. The September 1, 2026 run — which executed before the backlog events arrived in Snowflake later that morning — had completed successfully. Its incremental filter had selected events with event_timestamp > ‘2026-08-28 23:58:44’ (the MAX from the prior run) and had loaded the small number of August 29 events that had arrived before the Kafka outage. After that run, the MAX(event_timestamp) in stg_shipment_events had advanced to approximately 2026-08-29 01:47:23. The September 2 dbt run executed after the backlog had arrived. Its filter was: WHERE event_timestamp > ‘2026-08-29 01:47:23’. This filter DID select most of the August 29 backlog events (those with event_timestamp > 01:47:23), but NOT the August 29 events with event_timestamp between 2026-08-28 23:58:44 and 2026-08-29 01:47:23 (which had already been loaded by the September 1 run). Over successive runs through September 2–3, the MAX advanced through August 29, August 30, and August 31 as the backlog events were loaded. By September 4, all August 29–31 events appeared to have been consumed — but they had not.

-- The gap: events with event_timestamp < MAX(event_timestamp) at time of arrival

-- September 1 02:00 UTC dbt run:
-- MAX(event_timestamp) in stg_shipment_events BEFORE run: 2026-08-28 23:58:44 UTC
-- Filter: WHERE event_timestamp > '2026-08-28 23:58:44'
-- Records loaded: ~12,000 (Aug 28 events that arrived before Kafka outage)
-- MAX(event_timestamp) AFTER run: 2026-08-29 01:47:23 UTC
-- Backlog events NOT YET in Snowflake RAW at this point (arrive Sep 1 09:00+)

-- September 2 02:00 UTC dbt run (backlog now in Snowflake RAW):
-- MAX(event_timestamp) in target: 2026-08-29 01:47:23 UTC
-- Filter: WHERE event_timestamp > '2026-08-29 01:47:23'
-- Backlog events available in RAW: Aug 29-31, but only those AFTER 01:47:23 Aug 29
-- Events permanently missed: Aug 28 23:58:44 → Aug 29 01:47:23 UTC (~1,847 events)
--   These arrived AFTER the Sep 1 run had already set MAX past them
--   event_timestamp < MAX(target) → not selected by ANY future incremental run

-- Over Sep 2-4, MAX advances through Aug 29-31 as remaining backlog is loaded
-- But the events that arrived AFTER the MAX had advanced past their window
-- are permanently locked out: event_timestamp < current MAX → never selected
-- Total permanently excluded: 847,293 events across multiple missed windows

The mismatch occurred because the backlog events arrived in Snowflake in batches over multiple days, and each dbt run advanced the MAX(event_timestamp) target only through the events it had seen. Events in the Kafka connector’s consumer offset backlog were delivered to Snowflake in the order they were buffered in Kafka — but the Kafka Connect connector processed topic partitions in parallel, meaning events from different partitions arrived in Snowflake in a non-deterministic order. Some August 30 events arrived in Snowflake before some August 29 events (from different topic partitions). When the dbt run selected events WHERE event_timestamp > MAX(target), it selected the August 30 events (whose MAX was lower at that moment), loaded them, advanced MAX to August 30, and in so doing permanently locked out the August 29 events that were already in Snowflake RAW with timestamps before the new MAX.

Diagnosis: raw table vs. staging model event_id count by date

-- Diagnosis Step 1: Compare RAW vs. staging event counts by date
-- Snowflake, executed October 14 2026

SELECT
    DATE(event_timestamp)         AS event_date,
    COUNT(*)                      AS raw_row_count
FROM RAW.EVENTS.SHIPMENT_EVENTS
GROUP BY event_date
ORDER BY event_date;

-- Compare against:
SELECT
    DATE(event_timestamp)         AS event_date,
    COUNT(*)                      AS staging_row_count
FROM ANALYTICS.STAGING.STG_SHIPMENT_EVENTS
GROUP BY event_date
ORDER BY event_date;

-- Result: August 29-31 2026 discrepancy:
-- Date          RAW rows    Staging rows    Missing
-- 2026-08-29    291,847     184,293         107,554
-- 2026-08-30    312,293     231,847          80,446
-- 2026-08-31    289,293     135,000         154,293
-- (other dates: exact match)
-- Total missing from staging: 342,293 events

-- Step 2: Find event_ids in RAW but not in staging
SELECT COUNT(*) AS missing_events
FROM RAW.EVENTS.SHIPMENT_EVENTS r
WHERE NOT EXISTS (
    SELECT 1 FROM ANALYTICS.STAGING.STG_SHIPMENT_EVENTS s
    WHERE s.event_id = r.event_id
)
  AND DATE(r.event_timestamp) BETWEEN '2026-08-28' AND '2026-09-02';
-- Result: 847,293 events in RAW not in staging

-- Step 3: Check _fivetran_synced vs event_timestamp for missing events
SELECT
    MIN(event_timestamp)    AS min_event_ts,
    MAX(event_timestamp)    AS max_event_ts,
    MIN(_fivetran_synced)   AS min_fivetran_synced,
    MAX(_fivetran_synced)   AS max_fivetran_synced
FROM RAW.EVENTS.SHIPMENT_EVENTS r
WHERE NOT EXISTS (
    SELECT 1 FROM ANALYTICS.STAGING.STG_SHIPMENT_EVENTS s
    WHERE s.event_id = r.event_id
);
-- min_event_ts:      2026-08-28 23:58:47 UTC
-- max_event_ts:      2026-08-31 23:59:01 UTC
-- min_fivetran_synced: 2026-09-01 09:14:22 UTC  ← arrived AFTER dbt had advanced MAX
-- max_fivetran_synced: 2026-09-03 17:41:08 UTC
-- Confirms: events have correct domain timestamps but arrived late in Snowflake

-- Step 4: Confirm filter logic is the cause
SELECT MAX(event_timestamp) AS target_max
FROM ANALYTICS.STAGING.STG_SHIPMENT_EVENTS
WHERE DATE(event_timestamp) BETWEEN '2026-08-28' AND '2026-08-29';
-- 2026-08-29 01:47:23 UTC
-- The incremental filter for the Sep 2 run was: WHERE event_timestamp > '2026-08-29 01:47:23'
-- Events with event_timestamp between '2026-08-28 23:58:47' and '2026-08-29 01:47:23'
-- (107,554 events) arrived AFTER this MAX was set → permanently excluded

Diagnosis was complete in 2.0 hours. The root cause was the canonical incremental filter anti-pattern: using the domain event timestamp as the incremental watermark assumes that events arrive in Snowflake in the same temporal order as their domain timestamps. The Kafka broker migration had violated this assumption by delivering three days of events in a non-deterministic order over two days, advancing the dbt MAX watermark past events that had not yet arrived in Snowflake.

The fix: switch incremental filter to _fivetran_synced and run a full-refresh backfill

Part 1: Update the incremental model filter to use the ingestion timestamp. The fix changes the incremental watermark from the domain event_timestamp to the Snowflake ingestion timestamp _fivetran_synced. With this filter, the incremental model selects all events that arrived in Snowflake after the last run’s MAX(_fivetran_synced) in the target table — regardless of what domain timestamp the events carry. Late-arriving events with old domain timestamps are still captured on the next run after they arrive, because their _fivetran_synced value (when they actually entered Snowflake) is after the watermark.

-- models/staging/stg_shipment_events.sql (AFTER fix)

{{
    config(
        materialized='incremental',
        unique_key='event_id',
        incremental_strategy='merge',
        on_schema_change='sync_all_columns'
    )
}}

SELECT
    event_id,
    event_timestamp,
    event_type,
    shipment_id,
    carrier_id,
    _fivetran_synced,
    CURRENT_TIMESTAMP() AS dbt_loaded_at
FROM {{ source('raw_events', 'shipment_events') }}

{% if is_incremental() %}
    WHERE _fivetran_synced > (
        SELECT MAX(_fivetran_synced)
        FROM {{ this }}
    )
{% endif %}

-- _fivetran_synced is the Snowflake Kafka Connector write timestamp.
-- Using this as the watermark ensures late-arriving events (Kafka backlog,
-- connector restarts, source maintenance windows) are captured on the next
-- run after they arrive in Snowflake, regardless of their event_timestamp.
-- The unique_key=event_id MERGE prevents duplicates if an event arrives
-- twice (connector retry) or if a full-refresh overlaps with a partial run.

Part 2: Full-refresh backfill to load the missing 847,293 events. Updating the incremental filter fixed future runs but did not retroactively load the missing August 29–31 events. A full-refresh run was executed via dbt run --full-refresh --select stg_shipment_events. This replaced the entire staging table with a complete reload from the RAW source, including the 847,293 events that had been permanently excluded by the old MAX(event_timestamp) filter. After the full-refresh completed (14 minutes 23 seconds), downstream models int_shipment_lifecycle and fct_shipments were rebuilt via dbt run --select int_shipment_lifecycle fct_shipments.

-- Post-fix validation: confirm missing events are now present

SELECT
    DATE(event_timestamp) AS event_date,
    COUNT(*)              AS staging_row_count
FROM ANALYTICS.STAGING.STG_SHIPMENT_EVENTS
WHERE DATE(event_timestamp) BETWEEN '2026-08-28' AND '2026-09-02'
GROUP BY event_date ORDER BY event_date;

-- After full-refresh:
-- 2026-08-28: 273,847  (matches RAW)  ✓
-- 2026-08-29: 291,847  (matches RAW)  ✓  (was 184,293 before fix)
-- 2026-08-30: 312,293  (matches RAW)  ✓  (was 231,847 before fix)
-- 2026-08-31: 289,293  (matches RAW)  ✓  (was 135,000 before fix)
-- 2026-09-01: 284,847  (matches RAW)  ✓

-- Missing events now in staging: 0
SELECT COUNT(*) AS still_missing
FROM RAW.EVENTS.SHIPMENT_EVENTS r
WHERE NOT EXISTS (SELECT 1 FROM ANALYTICS.STAGING.STG_SHIPMENT_EVENTS s WHERE s.event_id = r.event_id);
-- 0  ✓

-- Downstream: fct_shipments delivery confirmation count for Aug 29-31
SELECT event_date, COUNT(*) AS delivered_count
FROM ANALYTICS.MARTS.FCT_SHIPMENTS
WHERE event_type = 'DELIVERED'
  AND DATE(event_timestamp) BETWEEN '2026-08-29' AND '2026-08-31'
GROUP BY event_date ORDER BY event_date;
-- 2026-08-29: 58,847   ✓  (was 37,293 before fix)
-- 2026-08-30: 62,293   ✓  (was 46,847 before fix)
-- 2026-08-31: 57,847   ✓  (was 27,000 before fix)

-- The 2,847 "unconfirmed" shipments in the carrier dispute report:
SELECT COUNT(*) AS dispute_shipments_now_confirmed
FROM ANALYTICS.MARTS.FCT_SHIPMENTS
WHERE event_type = 'DELIVERED'
  AND shipment_id IN (SELECT shipment_id FROM disputed_shipments_list);
-- 2,847  ✓ (all dispute shipments now show DELIVERED confirmation)

Total engagement time: 4.5 hours (2.0h diagnosis: RAW vs. staging row count comparison by date, missing event_id count query, _fivetran_synced vs event_timestamp analysis of missing events, incremental filter logic review and MAX watermark tracing through the September 1–4 dbt run history, Kafka broker migration timeline review with platform engineering team; 1.5h fix: incremental filter update in stg_shipment_events.sql, dbt full-refresh run for stg_shipment_events, rebuild of int_shipment_lifecycle and fct_shipments, post-fix row count validation by date, carrier dispute shipment confirmation validation; 0.5h dbt test additions: source freshness test for raw.events.shipment_events (error if MAX(_fivetran_synced) > 4 hours ago), event count comparison test comparing RAW and staging row counts by date (warning if staging missing > 0.1% of RAW rows for any date in last 14 days); 0.5h documentation and retainer SOP update).

Why the max(event_timestamp) incremental pattern is a recurring dbt retainer risk

The Meridian Freight Solutions incident illustrates the structural assumption embedded in the canonical dbt incremental filter: WHERE event_timestamp > (SELECT MAX(event_timestamp) FROM {{ this }}). This filter is correct when events arrive in the raw source table within seconds or minutes of their domain timestamp — which is true for most Kafka Connect, Fivetran, and Airbyte pipelines under normal operating conditions. The filter becomes incorrect whenever the ingestion pipeline experiences a delay of more than one dbt run interval, delivering events with past domain timestamps after the dbt watermark has already advanced beyond them.

The failure is entirely silent in dbt’s monitoring interface because it does not represent a dbt error: the model correctly loads all records that its filter selects. The missing records are simply not selected. No dbt test fails (unless a specific test for missing event IDs has been configured against the raw source). No dbt run fails. The Snowflake Query Profile shows normal execution. The only observable evidence is in the downstream business metrics — a gap in delivery confirmations that matches the Kafka outage window, discovered weeks later when carrier invoices don’t reconcile.

The fix — switching from event_timestamp to _fivetran_synced (or an equivalent ingestion-timestamp column) as the incremental watermark — is the correct long-term solution for any system where late-arriving records are possible. The unique_key MERGE strategy prevents duplicates if events are delivered twice. The tradeoff is that _fivetran_synced-based incremental models will re-process recent events on each run even if they haven’t changed (because new late arrivals from the ingestion pipeline will have _fivetran_synced values after the prior watermark). For high-volume event tables, this tradeoff should be evaluated against the full-refresh cost; for most retainer-scale analytics platforms, the incremental run overhead is acceptable.

How HourTab helps dbt developer retainers

The Meridian Freight Solutions engagement totaled 4.5 hours: diagnosis (2.0h — 0.5h running the RAW vs. staging event count comparison by date in Snowflake, identifying the August 29–31 discrepancy: 847,293 events present in RAW but absent from staging; 0.5h querying the missing event_ids for their _fivetran_synced vs. event_timestamp values, confirming they arrived in Snowflake on September 1–3 with domain timestamps of August 29–31; 0.5h reconstructing the incremental filter watermark trace through the September 1–4 dbt run history using the Snowflake QUERY_HISTORY table and the Airflow task log; 0.5h reviewing the Kafka broker migration timeline with the platform engineering team and confirming the consumer group rebalance window of August 29–31), fix (1.5h — 0.25h updating the incremental filter in stg_shipment_events.sql from event_timestamp to _fivetran_synced, committing and opening a pull request in the dbt project repository, reviewing with the data engineering team; 0.5h running dbt run --full-refresh --select stg_shipment_events (14m23s), monitoring Snowflake credit consumption during full refresh; 0.25h running dbt run --select int_shipment_lifecycle fct_shipments to rebuild downstream marts; 0.5h post-fix validation: per-date row count comparison of RAW vs. staging, missing event_id check returning 0, fct_shipments delivery confirmation counts for August 29–31 vs. carrier portal counts, 2,847 dispute shipments confirmed DELIVERED), and monitoring additions and documentation (1.0h — 0.5h adding a dbt source freshness test for raw.events.shipment_events with error_after 4 hours and a custom singular test comparing RAW vs. staging row counts by event_date for the last 14 days, warning if any date has staging missing > 0.1% of RAW; 0.5h post-incident write-up, retainer SOP updated to add quarterly incremental model filter audit for all incremental models, reviewing whether event_timestamp or _fivetran_synced is the correct watermark for each model based on each source system’s ingestion characteristics).

In a HourTab retainer log, the incident session reads: “stg_shipment_events incremental model: filter on MAX(event_timestamp) permanently excludes late-arriving Kafka backlog events; Kafka broker migration August 29–31 2026 creates 3-day event backlog; 847,293 events delivered to Snowflake RAW September 1–3 with event_timestamps of August 29–31; dbt MAX(event_timestamp) advances past them during September 1–4 runs before all backlog events arrive; events with event_timestamp < current MAX never selected by incremental filter; fct_shipments missing 847,293 delivery events for August 29–31; 2,847 shipments appear undelivered; carrier dispute reconciliation October 14 discovers gap; fix: change incremental filter to _fivetran_synced > MAX(_fivetran_synced FROM target); dbt run --full-refresh --select stg_shipment_events reloads all 847,293 missing events; downstream mart rebuild; 2,847 dispute shipments now confirmed DELIVERED; dbt source freshness + row count tests added; 4.5h.”

HourTab turns a time-tracker CSV into a public retainer-hours URL your client can bookmark — no login, no portal, no “how many hours do I have left?” emails. Free for one retainer. $9/mo for up to ten.

FAQ: dbt developer retainers

See also: Snowflake developer retainer · Power BI developer retainer · Azure Data Factory developer retainer