This version is still in development and is not considered stable yet. For the latest stable version, please use Korvet 0.20!

Redis Streams Storage

Korvet uses Redis Streams as its primary storage layer for all messages.

Why Redis Streams?

  • Low latency: Sub-millisecond read/write performance

  • Consumer groups: Built-in support for coordinated consumption

  • Persistence: AOF and RDB for durability

  • Scalability: Handle millions of messages per second

Stream Structure

In Redis-only mode (the default), each topic partition maps to a single Redis Stream. When tiered storage is enabled for the topic (remote.storage.enable=true), the partition is instead split into segment streams that seal and roll over as they fill.

For the exact stream-key patterns, segment keying, and record field layout, see Redis Data Structures.

Segments (tiered storage only)

Segments apply only to topics with tiered storage enabled; Redis-only topics use the single per-partition stream described above.

A tiered partition starts with a single open segment (segmentId 0). The storage worker seals the open segment and opens the next one (1, 2, …​) when the open segment reaches its message-count limit or its configured age. Sealing into discrete segments enables:

  • Efficient retention: Drop entire sealed segments when data expires

  • Archival: Offload sealed segments to remote storage without blocking writes to the open segment

  • Memory tiering with Redis Flex: Because each segment is a separate stream key, Redis Flex (Auto Tiering) can keep the open segment hot in RAM while transparently demoting colder sealed segments to flash/SSD — a warm tier between RAM and the remote object store. This is handled by Redis and needs no Korvet configuration.

Write Behavior

A few local-tier write specifics worth knowing:

  • The broker assigns each stream entry ID (a timestamp-based timestamp-sequence value); it is not the * auto-generated ID.

  • The XADD call sets only the entry ID and carries no retention arguments — retention is applied separately by the storage worker (see Retention Policies).

  • Null versus empty (for example a null-value tombstone versus an empty value) is encoded by stream-field presence: a field is omitted when its component is absent and present-but-empty when the component is a zero-length array.

The record field layout itself is documented in Redis Data Structures.

At-Rest Compression

Korvet can compress the value field before writing it to Redis. The codec is set per topic via storage.compression.type, falling back to the server-level korvet.storage.local.compression.codec (default none). The default stores values uncompressed and directly readable by non-Kafka clients. When a codec is used, the value is written as that codec’s standard frame format with no Korvet-specific marker prefixed (gzip, the LZ4 frame, the Snappy framing format, or a zstd frame), so a non-Kafka client can decompress the field with any stock decompressor for that codec — knowing only the topic’s codec.

storage.compression.type is fixed at topic creation. Because the stored value carries no codec marker, the effective codec is pinned at creation: a topic that does not set its own storage.compression.type snapshots the server-level default into its config, so a later change to the server default never reinterprets a topic’s already-written values.

When upgrading from a release whose server-level default was lz4, existing topics retain their pinned codec, including any explicitly chosen codec. Newly created topics without a topic or pattern override use the current server default (none unless configured otherwise). To keep using lz4 for new topics, set korvet.storage.local.compression.codec: lz4 or an explicit storage.compression.type when creating them.

Supported codecs are none, gzip, snappy, lz4, and zstd. Because Korvet stores each record as its own Redis Stream entry, the codec compresses one record value at a time. That makes value size the deciding factor for plain (non-dictionary) compression: it only pays off for values of roughly 1-4 KB and above. A single few-hundred-byte JSON log event compresses only marginally on its own (a ~240 B event reaches about 1.3x with gzip, and frame formats such as LZ4 add ~15 B of header/end-mark per record), even though the same events compress ~9x when batched — the redundancy in log streams lives across records, not within one, and a plain per-record codec never sees it.

Choosing a Codec

Pick by what the topic stores and what you optimize for:

Record profile Recommended codec Why

Structured records — JSON, logs, metrics, events — of any size up to a few KB

zstd

Best storage reduction at every size. For small records (below ~1 KB), the per-topic trained dictionary (on by default for zstd topics) recovers the cross-record redundancy that plain per-record compression cannot see.

Large compressible values (roughly 4 KB and above) on CPU-constrained brokers

lz4 or snappy

Lowest codec CPU per byte with a still-useful ratio. Prefer zstd when storage cost matters more than broker CPU.

Already-compressed or high-entropy payloads — images, encrypted data, compressed binaries

none

No codec helps at any size; compression only adds CPU cost.

Values that non-Kafka clients read directly from Redis (XRANGE) with no decompression step

none

Any codec requires readers to decompress; dictionary-compressed zstd additionally requires the dictionary blob.

Guidance by topic shape:

  • Small-event topics (few-hundred-byte records, such as log pipelines): use zstd. Zstd topics enable dictionary compression by default, which recovers the cross-record redundancy that plain per-record compression cannot see. Use none if you prefer to skip compression entirely, or must (see below).

  • Large-value topics (roughly 1-4 KB and above) with repetitive content — big JSON documents, telemetry blobs, text payloads: start with zstd for the best storage reduction, or snappy when CPU headroom is tighter and you want the lowest codec cost with good compression.

  • Random or already-compressed payloads (images, encrypted data, compressed binaries): use none — no codec helps, at any size.

  • Use none whenever non-Kafka clients must read values directly from Redis with XRANGE without a decompression step.

Zstd Dictionary Compression

For zstd topics, Korvet can train a per-topic dictionary from sampled record values and compress every subsequent record against it. This makes per-record compression effective for small records: the dictionary carries the redundancy that repeats across records, so each small value no longer has to supply its own.

Dictionary compression is controlled by the creation-only topic config storage.compression.dictionary.enabled, which is valid only when the effective storage.compression.type is zstd and defaults to true for zstd topics. While a new topic has no dictionary yet, the broker samples record values on the produce path (bounded by the server-level korvet.storage.local.compression.dictionary.* settings), trains a dictionary once the sample budget is reached, and publishes it to Redis under korvet:storage:zdict:<topic> so all brokers compress the topic with the same dictionary. If training or publication fails, the topic stays on plain zstd frames — records are always readable.

Stored values remain standard zstd frames: the dictionary id travels in the frame’s native Dictionary_ID header, so every value self-describes which dictionary decodes it. Records written before the dictionary was trained are plain zstd frames (dictionary id 0) and stay readable forever. A non-Kafka client can still decode dictionary-compressed values with any stock zstd decompressor, but it needs the dictionary blob (readable from korvet:storage:zdict:<topic>) in addition to the frame — if values must be readable with no decompression step at all, use none.

Measured Memory Savings

The table below shows effective compression ratios for zstd with dictionary compression: actual Redis memory consumed by the backing stream (MEMORY USAGE), measured end-to-end through the produce path against a local Redis instance. The ratio therefore includes everything an uncompressed topic would also pay — stream entry IDs, listpack/radix-tree structure, and the uncompressed key and timestamp fields — not just the codec’s output. Payloads are schema-homogeneous structured JSON (log/order-style events with repeated field names and value vocabulary), 2,000 records per size, produced after dictionary training completed.

Record size Redis memory per record (none) Redis memory per record (zstd + dictionary) Effective memory reduction

~512 B

624 B

176 B

3.5x

~1 KB

1,146 B

276 B

4.2x

~2 KB

2,197 B

466 B

4.7x

~4 KB

4,219 B

844 B

5.0x

~6 KB

6,313 B

1,228 B

5.1x

Read these as a realistic expectation for structured JSON: 3.5-5x less Redis memory across 512 B - 6 KB records. The dictionary matters most exactly where plain compression fails — at ~512 B, plain zstd manages only ~1.3x while the dictionary reaches 3.5x — and its contribution tapers as records grow large enough to carry their own redundancy. Heterogeneous topics (mixed schemas, or payloads dominated by unique IDs or base64 blobs) will land lower; measure on your own data before capacity planning.

Compression is a trade-off, not a free win: produce pays the compression cost before XADD, and fetch pays the decompression cost after XREAD. For compressible payloads, smaller stored values can more than offset that CPU cost because Redis writes and reads less data.

The following local benchmark uses a batch of 1,000 JSON records, each 1 KiB before compression. Write latency is compress 1,000 records + one pipelined batch of 1,000 XADD commands; read latency is XREAD COUNT 1000 + decompress 1,000 records. Values are median batch latencies from a local Redis instance and are intended as directional guidance, not end-to-end broker latency. The compressed write and read paths are faster in this JSON workload because the reduced Redis I/O more than pays for codec CPU time.

Codec Stored size Total write latency Write change vs none Total read latency Read change vs none

none

100%

29 ms

baseline

7 ms

baseline

lz4

30%

15 ms

48% lower

5 ms

34% lower

snappy

31%

13 ms

57% lower

4 ms

48% lower

zstd

22%

10 ms

64% lower

4 ms

49% lower

gzip

23%

15 ms

49% lower

5 ms

36% lower

For JSON-like payloads around this size — near the low end of where plain per-record compression starts to pay — snappy is the lowest-CPU choice, while zstd stores the least data and produced the lowest total write/read batch latency in this run. Below this size the picture inverts for plain codecs: small values (such as few-hundred-byte log events) gain little or nothing per record, so compression adds CPU cost without reducing Redis I/O enough to pay for it. For those topics, use zstd with dictionary compression (the default for zstd topics), which restores useful ratios on small records. Random and already-compressed payloads gain nothing from any codec at any size; leave those topics uncompressed.

At-rest compression is independent of Kafka protocol compression (compression.type). At-rest compression applies to data stored in Redis, while protocol compression applies to data in transit between Kafka clients and Korvet.

Changing the Codec on an Existing Topic

There is no in-place codec change. Both storage.compression.type and storage.compression.dictionary.enabled are creation-only settings; attempts to alter them are rejected with INVALID_CONFIG (<setting> cannot be altered after topic creation). Changing the server-level korvet.storage.local.compression.codec does not change the codec of existing topics, even for new writes: each topic keeps the effective codec pinned at its creation.

korvet migrate is not a general-purpose re-encoding command for current topics. It handles supported legacy layout upgrades, including rewriting 0.12.5 values with a selected target codec and converting leftover legacy committed-offset keys on the current layout. See Migrating for its supported sources and offline procedure.

To change a current topic’s codec, create a new topic and re-produce the retained data through the Kafka API. The following procedure pauses source writes for a bounded copy; it is not a zero-downtime or exactly-once migration recipe. Rehearse it with your copy tool and consumers before production.

  1. Choose and measure the codec. Follow Choosing a Codec. For an initial plain-codec comparison, sample the source topic:

    korvet topics stats orders --bootstrap-server localhost:9092 --no-interactive

    topics stats re-compresses sampled, decoded values with plain codecs. It does not use a trained zstd dictionary or measure actual Redis memory usage, even in its stored and ratio columns. For dictionary-enabled zstd, measure a representative test topic after dictionary training; see Zstd Dictionary Compression and Measured Memory Savings.

  2. Create an empty destination. Match the source partition count and review retention, cleanup policy, timestamp mode, tiered-storage settings, ACLs, and capacity for both copies. This example assumes a three-partition source and explicitly enables dictionary compression:

    kafka-topics --bootstrap-server localhost:9092 --create \
  --topic orders-v2 --partitions 3 --replication-factor 1 \
  --config storage.compression.type=zstd \
  --config storage.compression.dictionary.enabled=true

    Add the other settings needed for your workload; the example does not clone the source’s configuration. Choose storage.compression.dictionary.enabled=false at creation if you need plain zstd instead. Keep destination producers and application consumers stopped until verification.

  3. Freeze the cutover boundary. Pause all source producers and flush outstanding writes. Stop application consumers after their intended processing/checkpoint commits. Record each partition’s copy bounds and each group’s next unprocessed record or application checkpoint. Ensure retention and compaction cannot remove records needed for the copy or verification during the migration.

  4. Copy and verify. Use a Kafka consumer/producer pipeline tested with Korvet that reads the chosen retained range and writes to the matching destination partitions in order. Preserve keys, values (including null tombstones), headers, and timestamps where required. The destination broker applies its new at-rest codec; do not copy raw Redis entries or change stored topic metadata. Wait for destination acknowledgements and verify per-partition record identities and content against the frozen source bounds. Retrying a partially completed copy can duplicate records unless the tool tracks acknowledged progress and handles retries accordingly.

  5. Set consumer positions, then switch traffic. With application consumers still stopped, choose and verify each group’s destination starting positions as described below. Point producers and consumers at the destination, then resume them and validate processing. Retain the source and its checkpoints through an agreed rollback window; remove it only after verification and rollback planning. Once the destination has new writes, switching back to the old topic alone would omit those writes — rollback needs a reconciliation plan.

Offsets are not preserved by the copy. Korvet derives each record’s offset from its Redis Stream entry ID (see Topics, Partitions, and Offsets), and re-producing writes new entries with new IDs. Source offsets are not portable to the new topic; do not carry over numeric committed offsets or assume a fixed offset difference.

Either replay from the destination’s beginning with consumers that tolerate reprocessing, or use a verified per-partition mapping from each source processing checkpoint to the next destination record (for example, a mapping recorded by the copy tool from producer acknowledgements). Review the planned positions before applying them and keep group members stopped while changing offsets.

Timestamp-based resets are not an exact checkpoint mapping. They require preserved record timestamps and verified timestamp-search behavior; equal or out-of-order timestamps can select a different processing boundary. A destination using LogAppendTime reports the new append time, not the source record’s time. Do not use a timestamp reset as a lossless substitute for a verified record mapping. See Record Timestamps.

Retention Policies

Korvet enforces retention with a background storage worker that issues XTRIM, not at produce time:

  • retention.ms: Time-based retention (default: 7 days)

  • retention.bytes: Size-based retention (default: unlimited)

  • compression.type: Compression type for fetch responses (none, gzip, snappy, lz4, zstd)

How Retention Works

A leader-locked background storage worker periodically trims each stream with XTRIM:

  • Count-based (retention.bytes): XTRIM <streamKey> MAXLEN <count>. The byte limit is converted to a message count by dividing by the topic’s measured average message size (falling back to 1024 bytes when no measurement exists yet).

  • Time-based (retention.ms): XTRIM <streamKey> MINID <minTimestamp>, where minTimestamp is the current time minus retention.ms.

# Count-based trim
XTRIM korvet:storage:local:my-topic:0 MAXLEN 1000

# Time-based trim (keep messages newer than minTimestamp)
XTRIM korvet:storage:local:my-topic:0 MINID 1234567890000

  • Retention is enforced by the background storage worker, not at produce time. XADD carries no retention arguments.

  • XTRIM uses exact trimming (no ~ / LIMIT) for predictable retention behavior.

  • Size-based retention (retention.bytes) is converted to a message count using the topic’s measured average message size.

Configuring Retention

Set retention when creating topics via Kafka Admin API:

Example: Create topic with retention configuration
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");

AdminClient admin = AdminClient.create(props);
NewTopic topic = new NewTopic("my-topic", 3, (short) 1);
topic.configs(Map.of(
    "retention.ms", "86400000",      // 1 day
    "retention.bytes", "1073741824", // 1 GB
    "compression.type", "lz4"        // Compress fetch responses with LZ4
));
admin.createTopics(List.of(topic));

Or configure defaults in application.yml via a catch-all pattern:

korvet:
  topics:
    - name: "*"
      retention-time: 7d
      retention-bytes: 10GB
      compression: lz4

Manual Trimming

For manual stream management, use Redis’s XTRIM command:

# Trim by count (keep last 1000 messages)
XTRIM korvet:storage:local:my-topic:0 MAXLEN 1000

# Trim by age (keep messages newer than the given timestamp)
XTRIM korvet:storage:local:my-topic:0 MINID <timestamp-in-ms>

Performance Tuning

Async Operations

Korvet uses asynchronous Redis operations for maximum throughput:

  • All Redis commands use Lettuce’s async API (RedisFuture)

  • Operations return CompletableFuture to avoid blocking

  • Multiple operations execute in parallel

  • Netty event loop threads remain non-blocking

Benefits: * Higher throughput with fewer threads * Better resource utilization * Reduced latency under load

Pipelining

Korvet automatically batches Redis operations using Lettuce’s command pipelining:

  • Multiple commands are batched together

  • setAutoFlushCommands(false) delays command execution

  • flushCommands() sends all commands in a single network round-trip

  • Significantly improves throughput for high-volume producers

Example: Producing 1000 messages sends 1000 XADD commands in a single pipeline instead of 1000 round-trips. * Configurable pool size based on workload

Compression Types

Korvet supports two types of compression:

1. At-Rest Compression

Compresses the stored record value at rest in Redis Streams:

  • Where: Applied to the value field in Redis Streams

  • When: At write time (produce) and read time (fetch)

  • Configuration: Per topic via storage.compression.type (create-time only), falling back to the server-level korvet.storage.local.compression.codec (default none). The effective codec is pinned at topic creation. Set a topic to none to keep values directly readable by non-Kafka clients.

  • Use case: Reduce Redis memory usage for large or repetitive data

How it works:

  • Producer writes: The value is compressed before storing in Redis, as the topic codec’s standard frame format with no Korvet-specific marker

  • Consumer reads: The value is decompressed when fetching from Redis

  • The codec is pinned per topic at creation, so the value field is always a plain standard frame a non-Kafka client can decompress with any stock decompressor for that codec

  • Transparent to Kafka clients - they receive uncompressed data

  • Independent of Kafka protocol compression

Benefits:

  • Reduces Redis memory usage

  • Lower storage costs

  • Faster Redis persistence (smaller AOF/RDB files)

  • No impact on Kafka client compatibility

See At-Rest Compression for configuration details, and Changing the Codec on an Existing Topic for how to move an existing topic to a different codec.

2. Protocol Compression (Kafka Standard)

Compresses Kafka protocol messages between clients and Korvet:

  • Where: Applied to Kafka Fetch/Produce request/response payloads

  • When: During network transmission

  • Configuration: compression.type topic config (none, gzip, snappy, lz4, zstd)

  • Use case: Reduce network bandwidth between Kafka clients and Korvet

How it works:

  • Producer side: Kafka clients can send compressed or uncompressed batches; Korvet decompresses them before storing

  • Consumer side: Korvet compresses fetch responses based on topic’s compression.type configuration

  • Storage: Messages are stored uncompressed in Redis (unless at-rest compression is enabled)

Benefits:

  • Reduces network bandwidth for fetch responses

  • Transparent to clients - works with all Kafka clients

  • Flexible per-topic configuration

  • Standard Kafka feature

Compression Comparison

Feature At-Rest Compression Protocol Compression

Purpose

Reduce Redis memory usage

Reduce network bandwidth

Applied at

Redis storage layer

Kafka protocol layer

Configuration

korvet.storage.local.compression.codec

compression.type

Affects

Redis memory, persistence

Network traffic

Transparent to

Kafka clients

Storage layer

Recommended for

Large values (roughly 1-4 KB and up) with zstd, or snappy when CPU is tighter; small-event topics with zstd + dictionary (default for zstd)

High-throughput consumers
You can use both types of compression together! For example, set korvet.storage.local.compression.codec=zstd to save Redis memory on a large-value JSON topic and compression.type=lz4 for fast network compression. Note the two operate at different granularities: protocol compression compresses whole fetch-response batches (so it helps even for small records), while at-rest compression works one record value at a time.

See Compression for protocol compression configuration details.

Next Steps