Hudi Pipelines at 12.9M msg/s: Why Offset Lag Lies About Freshness

Offset lag counts records; it does not measure time. That distinction matters enormously at petabyte scale. Srikanth Mamidala, writing from Twilio's data engineering organisation, documents a production gap where standard consumer lag metrics — including Kafka's records-lag-max and Hudi's own kafkaDelayCount — reported healthy numbers while downstream analytics teams consumed data that was sometimes hours stale. The root cause was architectural: Apache Hudi Delta Streamer manages its own Kafka checkpoints, storing per-partition offsets inside .hoodie commit files on S3 rather than committing to Kafka's consumer group offset tracking. Tools like Burrow, which monitor consumer group offsets, had no visibility into whether Hudi had actually landed those records into the lake.

The pipeline processes over five trillion records monthly as of Q4 2025 across self-hosted Kafka clusters, peaking at 12.9 million messages per second on Cyber Monday 2025. At that throughput, a freshness SLA measured only in offsets is not a freshness SLA at all. As pipeline architecture increasingly drives AI system outcomes, instrumentation gaps like this propagate silently into model quality.

How the Checkpoint Walk-Back Works

HoodieStreamer embeds a deltastreamer.checkpoint.key in each commit's metadata. The format is a string like topicName,0:offset0,1:offset1, representing next-to-read offsets per partition — if Hudi committed through offset 1199 on partition 0, it stores 1200. The metrics reporter, an external observer requiring no changes to producers or existing pipelines, runs every fifteen minutes via EventBridge on EMR Serverless and executes the following steps:

1. Use the Apache Hudi SDK's HoodieTableMetaClient (accessing S3 via the Hadoop S3A filesystem layer already present in the Spark runtime) to fetch the active timeline.
2. Walk commits in reverse chronological order up to MAX_COMMIT_DEPTH (defaulting to 100) until a commit containing deltastreamer.checkpoint.key is found.
3. Assign a dedicated Kafka consumer (with enable.auto.commit=false and its own consumer group ID) to each partition, seek to the checkpoint offset, and poll with a 500-millisecond timeout.
4. Take the record with the earliest timestamp across all partitions — not the average, not the latest — since the oldest waiting message defines worst-case lag for SLA purposes.
5. Compute lag = currentTimestamp - record.timestamp(), floored at zero via Math.max(0L, ...), capped at seven days for inactive pipelines.

The walk-back in step 2 was not in the original design. During a migration overlap where a legacy S3-sourced pipeline and the new Kafka-sourced pipeline wrote to the same Hudi table simultaneously, the most recent commit frequently belonged to the legacy writer, which embedded no checkpoint key. The original algorithm defaulted to epoch zero (19700101000000000 in Hudi's timestamp string format), converting to millisecond zero and producing astronomically large, meaningless lag values. The fix was to skip commits without the key and report commit depth — how many commits back the algorithm searched — as a separate diagnostic metric, making a rising commit depth a leading indicator of pipeline trouble before lag thresholds are crossed.

Edge Cases That Break Naïve Implementations

Clock skew: Producer clocks that drift ahead produce negative raw lag values. The Math.max(0L, ...) floor prevents negative reporting, but sustained negative values before the floor are worth alerting on separately as a clock skew signal.

Missing Kafka timestamps: Kafka represents an absent producer timestamp with the sentinel value -1 (ConsumerRecord.NO_TIMESTAMP), not null. Computing lag against -1 produces a garbage metric. The reporter detects this explicitly and suppresses reporting for that pipeline rather than publishing a misleading number.

Empty partitions: If consumer.poll() returns no records after seeking to the checkpoint offset, the reporter reports zero lag and returns immediately. The 500-millisecond poll timeout bounds the wait unconditionally.

The overarching design principle: a missing metric is more useful than a hallucinated one. When no valid checkpoint is found within MAX_COMMIT_DEPTH, the reporter emits nothing, surfacing as a "No Data" alert rather than a fabricated value.

SLA Enforcement as a Ratio, Not a Binary

Each pipeline declares its freshness threshold via a slaInMinutes field in an onboarding YAML config. The default is 60 minutes for streaming pipelines and 1,440 minutes for batch pipelines. The reporter expresses SLA status as lagSeconds / slaThresholdSeconds, a ratio clamped to 0.0–1.0, where 1.0 represents full breach.

The graded ratio directly mirrors SRE error-budget burn-rate tracking: teams set a warning at 0.7 and a critical at 1.0 instead of reacting only after a threshold is already breached. No changes were required to any existing ingestion pipeline.

Offset lag and time-in-queue are complementary, not interchangeable; running only one leaves a blind spot the other exposes. For teams on Delta Lake with Structured Streaming, Spark stores Kafka offsets in the streaming checkpoint directory rather than table metadata, but the core algorithm transfers directly: find the last committed Kafka offset, seek to it, compute the timestamp delta. Mamidala notes that Hudi's Iceberg migration path and integration with anomaly-detection libraries like Prophet and Luminaire are the next evaluation targets.