[Feature Request] Optional NVMe Hot-Tier Tail Cache (Lazy Read-Through) for Sub-3 ms Fetches

[pintu4146]

Who is this feature for?

Teams with < 5 ms latency SLO on fresh events—trading, ride-dispatch, gaming, ad-tech—who run ≥2 (configurable) consumer-groups and still want AutoMQ’s stateless S3 cost model.

What problem are they facing today?

Problem statement

Today any consumer fetch that targets offsets already flushed to S3 must perform an object-store GET. S3 p99 latency is 8–12 ms, so:

• Hot paths with sub-5 ms SLO (pricing, HFT, gaming) routinely violate latency budgets.
• When ≥ 2 (configurable) consumer-groups read the same tail segment, each pays the full S3 cost—doubling latency and GET charges.
• Rebalances, task migrations, or seek() retries trigger “thundering-herd” S3 bursts, causing lag spikes and SlowDown throttling.
• existing in‑memory LogCache/BlockCache (which typically holds only a few hundred MB and seconds of traffic

Because brokers are stateless, there is no on-node cache to serve these repeat reads at disk speed; the only choices are accept latency or shorten retention (losing replay).

Why is solving this impactful?

Impact

• Unblocks new use-cases – drops tail-read p99 from 8-12 ms to ≈ 2-4 ms, so latency-critical workloads (pricing, gaming, ad-tech) can stay on AutoMQ instead of migrating to stateful SSD-tiered forks.
• Cuts S3 bill & throttling risk – one cache hit avoids every subsequent GET for that segment, typically halving request cost and smoothing “thundering-herd” spikes during rebalances.
• Preserves AutoMQ’s differentiator – keeps brokers stateless; cache is disposable and off by default, so existing low-cost users aren’t forced to pay for SSDs.
• Small, opt-in change – implemented behind cache.type flag; no impact on producer ACK flow or durability model.
• Community value – adds a missing performance knob requested by several users, showcasing AutoMQ’s extensibility and encouraging further open-source contributions around storage-layer plugins.

Proposed solution

Proposed solution — “automq-hot-tail-cache” (Phase I)

• Lazy read-through NVMe cache

  • On the first S3 miss, broker writes the fetched segment to a bounded NVMe LRU cache; every later read of that segment is served locally (≈ 0.3 ms).

• Pluggable SPI

  • New interface SegmentCache (lookup, put, invalidate, stats).
  • Reference impls: InMemoryLRUCache (unit-test / CI) and NVMeFileCache (direct-I/O).

• CacheManager & eviction loop

  • Size + age caps (cache.max.size.bytes, cache.max.segment.age.ms) enforced by a background thread; ref-count protects in-flight reads.

• Zero write-path changes

  • Producer ACK still happens right after WAL fsync; cache population is post-read, so brokers remain stateless.

• Config & metrics (feature-flag OFF by default)

  • cache.type=none|nvme (default none) plus sizing knobs.
  • Micrometer metrics: cache_hit_rate, cache_evictions_total, cache_size_bytes, cache_read_latency_ms{quantile}.

• Incremental delivery

  1. PR-1 — SPI + in-mem cache + unit tests.
  2. PR-2 — NVMeFileCache, wiring in FetchService, metrics, integration test.
  3. Later RFC can add an eager write-through CacheReplicator and SPDK optimisations if the lazy mode proves valuable.

This keeps current clusters untouched, gives latency-sensitive users a drop-in speed boost, and preserves AutoMQ’s stateless scaling model.

Additional notes

RFC: Hot‑Tier NVMe Cache for AutoMQ Tail Segments Keeping brokers stateless

1  Motivation

AutoMQ’s shared‑storage architecture achieves massive cost savings by persisting log segments directly to object storage (e.g. Amazon S3) after a small fsync to a local WAL. While this design keeps brokers stateless and eliminates cross‑AZ replication, it introduces a latency gap for tail reads after data scrolls out of the existing in‑memory LogCache/BlockCache (which typically holds only a few hundred MB and seconds of traffic):

• p99 end‑to‑end latency ≈ 8–12 ms for consumers that fetch offsets already flushed to object storage (GET latency + deserialisation).
• Workloads such as financial tick feeds, gaming leaderboards, or micro‑services with sub‑5 ms SLOs cannot tolerate that penalty.

The goal of this RFC is to add an optional hot‑tier NVMe cache that preserves AutoMQ’s statelessness while targeting a tail‑read p99 below 3 ms (stretch goal < 2 ms), subject to empirical benchmarking.

1.1  Why a lazy cache adds value even with “consume‑once” semantics

Real‑world Kafka (and AutoMQ) deployments seldom read each record exactly once at the cluster level. Scenarios that trigger multiple tail reads — where a hot‑tier cache prevents S3 RTT on every replay — include:

Scenario	Why the same bytes are fetched again
Multiple consumer‑groups	ETL, monitoring, ML‑feature pipelines each have their own group ⇒ the newest 1‑2 segments are re‑fetched by N groups within seconds.
Rebalance / fail‑over	On partition reassignment the new consumer seeks to an older offset to avoid gaps; it re‑reads the tail segments it just saw on the old instance.
ksqlDB / Streams warm‑up	Stateful tasks replay the tail to rebuild in‑memory windows when they migrate between brokers.
Retry storms	An exception in processing triggers seek(offset‑1) on a batch of messages → immediate re‑reads of the same segment.
Analytics side‑loads	Ad‑hoc queries, Spark/Flink batch jobs or ML backfills often rewind a few minutes to gather context.

Example

Ride‑sharing platform: topic trip-events (1 KB msgs, 20 k msg/s). Consumer‑groups: pricing, fraud, driver‑incentives. A broker restart triggers a rebalance → each group re‑fetches the latest two log segments (~64 MiB). With lazy cache:

• First group incurs the S3 GET; segments enter NVMe.
• Second and third groups hit NVMe ⇒ save ≈ 8 ms × 20 k msg/s × 2 groups ≈ 320 ms per second of wall‑clock latency and ~800 GET requests.

Even when individual records are processed once per group, the same segment may be read multiple times across groups or recovery events. Caching the segment after the first miss converts all subsequent reads into sub‑millisecond NVMe hits while keeping the write path untouched.

2  Goals & Non‑Goals

Goals	Non‑Goals
* Provide a pluggable, bounded‑size cache for the most recent log segments.	* Re‑introduce follower replicas or ISR quorums.
* Keep brokers stateless (cache treated as disposable).	* Replace existing WAL mechanism.
* Maintain exactly‑once semantics and current producer ACK flow.	* Cache optimisation for random historical reads (cold scans).

3 Design Overview

# Lazy read‑through (Phase I)
Producer ──> WAL (fsync, ACK) ──> S3Uploader ──> S3 Object Store
                   │
              
           HotTierCache (NVMe) <─ populated on first consumer miss


Consumer Fetch
   │
   ├─> LogCache      (hit? 30 µs)          -- DRAM
   │
   ├─> NVMe Cache    (hit? 0.3 ms)         -- new layer
   │
   ├─> WAL tail      (if still resident)    0.5 ms
   │
   └─> S3 GET        (miss)                8–12 ms
           └─> put() into NVMe



Consumer Fetch Flow (lazy mode):
1. Check HotTierCache  -->  hit  --> return bytes
2. else read from S3   -->  update cache, populate HotTierCache

3.1 Key Components

Component	Responsibility
SegmentCache SPI	Simple interface (lookup, put, invalidate, stats).
InMemoryLRUCache	Default implementation for unit tests & CI.
NVMeFileCache	Production impl using direct‑I/O mapped files on local NVMe/EBS gp3.
CacheManager	Orchestrates eviction (size‑/age‑based) and populates cache on read path.

4 Data to Cache

• Complete log segments once they reach segment.bytes but before S3 upload.
• Optionally the current active segment tail via a memory‑mapped view of the WAL.
• Metadata: base‑offset → local‑file‑path, length, lastAccessNanos.

5  Interaction with Existing Storage Layer (lazy read‑through first)

• Write path (unchanged). Producer ACKs immediately after WAL fsync; only the existing S3Uploader streams segments to object storage. No second copy is created at write time.

• Read path (lazy cache fill):

  1. Consumer fetch → CacheManager.lookup(segmentId).
  2. Hit ⇒ zero‑copy read via FileChannel.transferTo (NVMe or memory file).
  3. Miss ⇒ delegate to S3StreamReader; once bytes are returned, CacheManager.put() stores the segment (or segment slice) in the cache asynchronously if cache.warmup=true.

• Eviction: background thread runs every evict.interval.ms to enforce cache.max.size.bytes and cache.max.segment.age.ms.

Future option: A follow‑up RFC could introduce an eager write‑through (CacheReplicator) path that copies segments to NVMe in parallel with the S3 upload. That optimisation is out of scope for Phase I to keep risk and complexity low.

6  Configuration Parameters (new)

Property	Default	Description
cache.type	memory | nvme | none	Global switch.
cache.max.size.bytes	20g	Soft cap; eviction brings size ≤ cap.
cache.max.segment.age.ms	600000	TTL for a segment in cache.
cache.warmup	false	If true, populate cache on S3 miss; else only writes.

7  Metrics & Observability

Metric	Type	Purpose
cache_hit_rate	gauge	Success ratio of cache lookups.
cache_evictions_total	counter	Tracks forced evictions.
cache_size_bytes	gauge	Current bytes on disk.
cache_read_latency_ms (p95/p99)	timer	Compare against S3 read latency.

8  Failure Handling

• Cache corruption ⇒ auto‑invalidate segment & fall back to S3.
• NVMe volume loss ⇒ broker restarts with empty cache (stateless guarantee). No data loss.
• Low‑disk‑space alert ⇒ force eviction to 80 % of cache.max.size.

9  Implementation Plan

1. Phase I (this RFC):

   • Introduce SegmentCache SPI + InMemoryLRUCache.
   • Implement NVMeFileCache using direct‑I/O mapped files.
   • Integrate lazy read‑through fill logic on the consumer fetch path.
   • Ship behind feature flag cache.type=nvme (defaults to none).

2. Phase II (future RFC):

   • Optional eager write‑through (CacheReplicator) pipeline.
   • Advanced eviction heuristics (LFU, adaptive age).
   • NUMA/SPDK optimisations.

3. Phase III: Empirical benchmarks (JMH, 128 partitions, 1 KB msgs, 10 Ki msgs/s).

4. Phase IV: Production‑grade metrics, docs, Helm/Compose examples.

10  Trade‑offs

Benefit	Drawback
p99 read latency drops from ~10 ms → ~2 ms.	Slight compute overhead (cache lookup, eviction).
Keeps brokers stateless (cache == disposable).	Adds local‑disk requirement; can’t run on truly disk‑less Fargate.
Zero data‑copy on rebalance (stateless property intact).	Cache warms up after scale‑in/out; first fetches may still hit S3.

11  Alternatives Considered

1. Increase WAL size to 20 GiB – wastes EBS and does not help segments already flushed.
2. Client‑side cache – complicated, per‑consumer, does not help fan‑out.
3. Classic tiered storage (SSD + S3) – re‑introduces stateful brokers; violates AutoMQ core principle.

12  Open Questions

1. Should we use Direct‑I/O or page‑cache for NVMeFileCache reads?
2. Eviction policy – pure LRU vs LFU? Mixed‑age LFUDA?
3. Config‑driven S3 prefetch on cache miss – is warm‑up desirable or too costly?
4. Any security implications of mapping files outside data directory?
5. Benchmarks: What latency target constitutes “success” for merge?

13  References

• AutoMQ documentation: Shared Storage architecture (2024)
• Confluent Kora paper: “Tiered Storage in Cloud‑Native Kafka” (SoCC 2022)
• Linux kernel mailing‑list discussion: O_DIRECT & posix_fadvise best practices

14  Call for Discussion & Next Steps

1. API surface — is the SegmentCache SPI minimal yet future‑proof (e.g., supports adaptive LFU)?
2. Eviction policy — stick with LRU or add LFU/age hybrid before merge?
3. Latency target — agree that p99 < 3 ms is “good enough,” or push for sub‑2 ms stretch goal?
4. Default state — should cache.type default to none (current proposal) or nvme in cloud‑native charts?
5. Benchmark acceptance criteria — what traffic pattern & hit‑rate threshold must Phase I demonstrate?

Please leave comments on each point or suggest additional concerns. Once consensus forms, we’ll convert this RFC into Phase I PRs (SPI + in‑mem cache) and a follow‑up PR for NVMeFileCache. Pull‑request series tracked under milestone “Hot‑Tier Cache”.

[SCNieh]

@pintu4146 Hi, thanks for your proposal and comprehensive documentation on the report, just a few questions and clarifications:

1. is the sub-5ms SLO requirement an end-to-end latency (from msg being produced to being consumed), or a first-byte-latency (latency of the first consume request) to the consumer?

• if it's an end-to-end latency, then I'm assuming that it's a scenario that requires messages being consumed in milliseconds since it has been produced - which is what we called a typical tail-read case, and in that case we're assuming the size of LogCache should be enough to preserve all the hot data. A few hundred MBs of LogCache should be enough to maintain the data "hot" for seconds with a produce throughput of also a few hundreds of MB/s, so the consumer should be able to consume the data directly from memory and ensure sub-5ms latency.
• if it's a first-byte-latency case, then adding a SSD cache can indeed decrease the consume latency, however I'm just wondering why first-byte-latency matters in this case, since when the data is evicted from the memory and consumer needs to fetch data from S3 or NVMe device, the end-to-end latency should already be couple of seconds, which should be unacceptable to any real-time events.

2. the BlockCache is optimized for multiple consumer groups reading from the same offset, data read from S3 will be cached in BlockCache so ideally there will be only "one" S3 GET request and the rest fetch requests will hit the cache

[pintu4146]

Hi @SCNieh,

Thanks so much for your detailed feedback and excellent questions! This is incredibly helpful for refining the proposal, and I appreciate the opportunity to clarify my perspective.

1. The Sub-5ms SLO & Defining the NVMe Cache's Sweet Spot

You're spot on: the sub-5ms end-to-end (E2E) latency SLO is the domain of tail-read scenarios, where LogCache is the star. With a few hundred MBs, it's designed to keep hot data in memory for those critical seconds, ensuring consumers meet this SLO.

My NVMe cache proposal is not intended to interfere with or replace LogCache for this vital hot path.

Instead, the NVMe cache shines by targeting performance improvements primarily for "warm data" scenarios:

• Data just aged out of LogCache: Due to RAM limits or high volume, this could be data seconds to minutes (or even up to an hour) old.
• Consumer Catch-Up: For consumers restarting, scaling, or those that are naturally slower processors.
• Near Real-Time Analytics & Reprocessing: Tasks needing frequent access to recent historical data that exceeds LogCache's capacity.

In these "warm tier" situations, you're correct that the original E2E latency (from produce to that specific consume event) will likely be longer than the sub-5ms real-time window. However, this is where First-Byte Latency (FBL) from the broker's storage system becomes paramount.

Why FBL for Warm Data is a Game Changer:
Significantly improving FBL via an NVMe cache (e.g., aiming for 5-15ms FBL from NVMe, versus 50-100ms+ for an S3 fetch if BlockCache also misses) can drastically boost the overall throughput of:

• Consumer catch-up processes.
• Reprocessing jobs.
• Analytical queries on recent data.

Even if a message's initial E2E journey isn't sub-5ms, fast access to it later as "warm" data makes these subsequent operations much more efficient.

2. NVMe Cache: Complementing BlockCache with Superior Cost-Effectiveness

I fully agree that BlockCache is crucial for optimizing S3 reads by caching data in RAM after the initial fetch, especially for multiple consumer groups. The NVMe cache proposal is designed to complement and supercharge this capability, specifically by tackling the limitations around RAM capacity and its associated costs when dealing with large "warm" datasets.

The core motivation is to offer a larger, more cost-effective caching tier than achievable by solely expanding RAM for BlockCache.

Let's illustrate with a practical scenario why customers might be forced into costly, oversized broker instances just for RAM:

Scenario: The 2 TB "Warm" Dataset Challenge

• Imagine an application frequently accessing a 2 TB "warm" dataset per AutoMQ broker.
• The broker's actual message processing workload (CPU/network for active writes and tail reads) is moderate, perhaps efficiently handled by an instance with 16 vCPUs.

Current Approach (Relying only on RAM-based BlockCache for this 2 TB):

1. RAM Needed: To cache 2 TB effectively in BlockCache, the instance needs well over 2 TB of total RAM (for BlockCache, OS, JVM, LogCache, etc.).
2. Forced Instance Choice: This compels the selection of a very large, memory-optimized cloud instance (e.g., on AWS, an x2iedn.16xlarge with 2048 GiB RAM also brings 64 vCPUs).
3. The Cost Issue – Paying a Premium for Underutilized CPU:
   The customer is now paying a significant premium for a 64 vCPU instance primarily to get the 2 TB of RAM. If their actual compute needs are only for 16 vCPUs, a large portion of the instance's cost is for CPU capacity they don't need and won't fully utilize. They're effectively forced into an expensive, oversized compute machine just for the RAM for the cache.

Proposed NVMe Cache Approach – Smarter Scaling:

1. Right-Sized Compute: The customer could select an instance appropriately sized for their 16 vCPU processing needs (e.g., an AWS r6id.4xlarge with 16 vCPUs, 128 GiB RAM, and ~1.8 TB of local NVMe, or a similar general-purpose instance with a large attached NVMe volume).
2. Efficient Cache Allocation:
   • The instance's RAM handles OS, JVM, LogCache, and a smaller, dynamic BlockCache for the absolute hottest blocks.
   • The 2 TB "warm" dataset primarily resides in the NVMe cache tier, which is significantly more cost-effective per GB than instance RAM.
3. ** The Cost Benefit – Optimized Spending:**
   The hourly cost for the compute-appropriate instance plus the NVMe storage would likely be substantially lower than the cost of the massive RAM-only instance. This strategy avoids paying for extensive, underutilized CPU capacity.

Essentially, the NVMe cache acts as a larger L2 cache to S3, sitting strategically between the RAM caches (LogCache/BlockCache) and object storage. It would absorb more reads for "warm" data before hitting S3, especially if the warm working set is larger than what can be economically held in BlockCache (RAM), or if BlockCache experiences high churn.

TL/DR: Key Goals of the NVMe Cache Proposal

• Boost Performance for "Warm" Data: Significantly improve First-Byte Latency and overall throughput for reads crucial for consumer catch-up, reprocessing, and recent historical analytics.
• Enhance Cost-Effectiveness: Offer a more economical solution for caching large "warm" datasets compared to scaling broker instance RAM to very high levels, thereby avoiding the over-provisioning of expensive CPU resources.
• Reduce S3 Load & Costs: Minimize frequent S3 GET requests (and associated costs) for this active warm tier.

I hope this detailed explanation clarifies the intended positioning and the cost-performance benefits I envision for an NVMe cache tier.

Thanks again for your insightful questions and the opportunity to elaborate , please let me know in case any further query !

[SCNieh]

@pintu4146 Thanks for the clarification, so the primary problem that this feature trying to resolve is frequent & small access to warm data, and that is indeed the scenario the current architecture does not optimized for. After having some discussion with other maintainers of the community, we all agree that it's of great meaning to add a warm cache layer to the current architecture. And here's some of our suggestions regarding your previous implementation proposal.

Instead of adding a new kind of cache, we prefer to enhance the current implementation of ObjectStorage, given that we already have an implementation of LocalFileObjectStorage, which is able to r/w to the local filesystem just like to the object storage. It maybe better to implement another abstraction of ObjectStorage that is able to achieve following goals:

1. every writes go into the remote object storage, and also the local filesystem, with the same object/file structure
2. every reads will attemp to read from local filesystem first, and then from the remote if missed

In that way the change to the overall codebase will be minimized and all we need to do is to choose the implementation of ObjectStorage based on configuration on startup

[pintu4146]

@SCNieh , Thanks for positive feedback and thoughtful discussion with other maintainers.
i have suggested both approach in the initial RFC

1. lazy read-through in case of miss fetch from object storage and populate NVMe cache for consumer.

2. eager write-through( async write to NVMe and object storage)
   ( which you have mentioned)

Lets have some discussion around both in terms further design phrases.
My plan was lazy-read through, please feel free to drop your suggestions.

[SCNieh]

@pintu4146
The lazy cache approach is good for storage effectiveness though, but it maybe hard to implement since you need to maintain a dedicated index from stream offset to the NVMe cache (since the data read from objecet storage can be just a part of an object), and the current indexing system in BlockCache cannot be reused, since the file structure in NVMe in this case is very different from what's in the objecet storage.

However, with write-through to both object storage and NVMe device, the file structure can be keeped the same between these two difference storage systems, thus the current indexing system can be reused. Take the "2TB" challenge you mentioned above, with this approach, the NVMe will always stores the most recent 2TB of warm data and makes any subsequent "warm" fetch request response in milliseconds and without triggering any S3 request.