For the latest stable version, please use Korvet 0.19!

Topic Management

This guide covers creating and managing topics in Korvet.

Topic defaults — including auto-create — are configured via the pattern list under korvet.topics. Each entry’s name is a glob pattern matched against topic names; entries are evaluated in declared order and combined first-match-wins per field.

Creating Topics

Automatic Topic Creation

By default, topics are not automatically created when you first produce to them or request metadata for them. Auto-creation is configured per pattern under korvet.topics:

korvet:
  topics:
    - name: "*"
      auto-create: false  # Enable/disable automatic topic creation (default: false)
      partitions: 1       # Default partitions for auto-created topics (default: 1)

When auto-creation is disabled, you must explicitly create topics before using them.

Explicit Topic Creation

You can create topics explicitly with standard Kafka tooling or with the bundled korvet-cli.

Using `korvet-cli topics`

korvet-cli topics mirrors kafka-topics syntax. It is a thin wrapper over the Kafka AdminClient and passes --config key=value pairs through verbatim. The only Korvet-specific topic config is offset.sequence.bits; all other accepted keys are standard Kafka topic configs such as retention.ms and segment.ms.

korvet-cli topics --bootstrap-server localhost:9092 \
  --create \
  --topic my-topic \
  --partitions 3 \
  --config retention.ms=604800000 \
  --config offset.sequence.bits=14 \
  --config segment.ms=3600000

--replication-factor is accepted for compatibility but ignored, as Korvet uses Redis for storage and replication.

Using `kafka-topics`

Upstream Kafka CLI tooling works for standard Kafka topic configs:

kafka-topics --bootstrap-server localhost:9092 \
  --create \
  --topic my-topic \
  --partitions 3 \
  --replication-factor 1

Creating Topics with the `offset.sequence.bits` Configuration

The upstream kafka-topics CLI performs client-side validation and rejects the Korvet-specific offset.sequence.bits config with an Unknown topic config name error.

Use korvet-cli topics (a thin AdminClient wrapper that does not validate config names client-side) when you want to set offset.sequence.bits, or use the Kafka AdminClient API directly.

To set offset.sequence.bits with upstream Kafka tooling, use the Kafka AdminClient API, which does not perform client-side validation:

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

try (AdminClient admin = AdminClient.create(props)) {
    NewTopic topic = new NewTopic("my-topic", 3, (short) 1);
    topic.configs(Map.of(
        "offset.sequence.bits", "14",
        "retention.ms", "604800000"
    ));
    admin.createTopics(List.of(topic)).all().get();
}

Accepted topic configurations:

The broker accepts only the following topic config keys. Any other key (including value.type and storage.compression) is rejected with INVALID_CONFIG ("Unknown or unsupported topic config").

retention.ms - Total time-based retention in milliseconds (across all tiers)
retention.bytes - Total size-based retention in bytes
segment.ms - Duration of each local stream bucket in milliseconds. Optional. Must be positive and less than the effective local retention window.
segment.bytes - Size of each local stream bucket in bytes
compression.type - Compression for Kafka fetch responses (none, gzip, snappy, lz4, zstd). Default: none
cleanup.policy - delete (retention-based trimming, the default), compact (key-based log compaction), or compact,delete (both). See Log Compaction.
offset.sequence.bits - Bits reserved for the per-millisecond sequence component in Korvet offsets. Range: 1-16. Default: 14. Settable only at topic creation; cannot be altered.
storage.compression.type - Codec for compressing the record value at rest in Redis (none, gzip, snappy, lz4, zstd). Unset inherits the server-level korvet.storage.local.compression.codec. Setting it to none stores the value uncompressed so non-Kafka clients can read it directly via XRANGE. Settable only at topic creation; cannot be altered.

Tiered storage configurations (when remote storage is enabled at server level):

remote.storage.enable - Enable tiered storage for this topic (Kafka KIP-405). Default: false
local.retention.ms - Time to keep in the local tier before Redis data expires. -2 = use retention.ms (Kafka KIP-405)
local.retention.bytes - Size to keep in the local tier before Redis trimming falls back to retention.bytes. -2 = use retention.bytes (Kafka KIP-405)

At-rest compression of the record value is set per topic via storage.compression.type (falling back to the server-level korvet.storage.local.compression.codec, default lz4). This is distinct from the Kafka-facing compression.type, which only affects fetch-response compression.

Log Compaction

Topics created or altered with cleanup.policy=compact (or compact,delete) are compacted by key: a background pass on the storage worker periodically deletes every record that has been superseded by a newer record with the same key, keeping only the latest record per key. This supports keyed-state topics such as Schema Registry journals and Kafka Connect config/offset topics, which rebuild their state by replaying a compacted topic.

Because Korvet derives Kafka offsets from Redis stream entry IDs rather than positions, compaction deletes superseded entries in place (XDEL): surviving records keep their original offsets, and consumers simply observe offset gaps — the same behavior as Kafka compaction.

Semantics:

compact-only topics ignore retention.ms/retention.bytes: the head of the log is never trimmed, only superseded keys are removed. compact,delete applies both compaction and retention trimming.
Records produced to a compacted topic must have a key; unkeyed records are rejected with INVALID_RECORD (standard Kafka behavior).
Tombstones (records with a null value) are retained as the latest record for their key so replaying consumers observe deletions. Tombstone purging (delete.retention.ms) is not implemented; tombstones are kept indefinitely.
min.compaction.lag.ms/max.compaction.lag.ms are not supported; compaction runs at the storage worker’s tick interval. Records appended while a compaction pass is running are left for the next pass.
cleanup.policy=compact cannot be combined with remote.storage.enable=true: compaction is not supported on tiered topics, and the combination is rejected at create/alter time.

Listing Topics

List all topics:

kafka-topics --bootstrap-server localhost:9092 --list

Describing Topics

Get details about a topic:

kafka-topics --bootstrap-server localhost:9092 \
  --describe \
  --topic my-topic

Deleting Topics

Delete a topic:

kafka-topics --bootstrap-server localhost:9092 \
  --delete \
  --topic my-topic

Altering Topic Configuration

Topics can be configured with:

Partitions: Number of partitions for parallelism (set during creation only)
Retention: Time-based (retention.ms) and size-based (retention.bytes) retention policies
Protocol Compression: Compression for Kafka fetch responses (compression.type)
At-Rest Compression: Codec for the record value stored in Redis (storage.compression.type, create-time only)
Offset Encoding: Per-topic offset sequence width (offset.sequence.bits, create-time only)
Bucketing: Time-bucketed local streams (segment.ms, segment.bytes)

Using kafka-configs CLI

Use korvet-cli topics --alter or kafka-configs to alter topic configurations.

korvet-cli topics --bootstrap-server localhost:9092 \
  --alter \
  --topic my-topic \
  --config retention.ms=604800000 \
  --config segment.ms=1800000

kafka-configs --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --alter \
  --add-config retention.ms=604800000,compression.type=lz4

offset.sequence.bits cannot be altered after topic creation; it is settable only at creation time.

Using AdminClient API

Alternatively, use the AdminClient API:

ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, "my-topic");
List<AlterConfigOp> ops = List.of(
    new AlterConfigOp(new ConfigEntry("retention.ms", "604800000"), AlterConfigOp.OpType.SET),
    new AlterConfigOp(new ConfigEntry("compression.type", "lz4"), AlterConfigOp.OpType.SET)
);
admin.incrementalAlterConfigs(Map.of(topicResource, ops)).all().get();

See Redis Streams storage for details on how records are stored.

Describing Topic Configuration

View current topic configuration using korvet-cli topics --describe or kafka-configs --describe:

korvet-cli topics --bootstrap-server localhost:9092 \
  --describe \
  --topic my-topic

kafka-configs --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --describe

Protocol compression types (compression.type):

none - No compression (default)
gzip - Good compression ratio, higher CPU usage
snappy - Balanced compression and speed
lz4 - Fast compression, lower CPU usage
zstd - Best compression ratio, moderate CPU usage

At-rest compression in Redis is set per topic via storage.compression.type (create-time only), falling back to the server-level korvet.storage.local.compression.codec (default lz4). It is independent of the Kafka-facing compression.type, which only affects fetch-response compression. Set storage.compression.type=none to keep values directly readable by non-Kafka clients.

See Compression for more details on protocol compression.

Tiered Storage Configuration

When tiered storage is enabled at the server level, you can configure per-topic retention policies to control when data moves between tiers.

Configuring Tiered Storage with AdminClient API

Use the AdminClient API to configure tiered storage (since kafka-configs --alter is not supported):

NewTopic topic = new NewTopic("my-topic", 3, (short) 1);
topic.configs(Map.of(
    "remote.storage.enable", "true",
    "retention.ms", "31536000000",           // 1 year total
    "local.retention.ms", "86400000"         // 1 day in local tier
));
admin.createTopics(List.of(topic)).all().get();

This configures:

Local tier: 1 day (local.retention.ms=86400000)
Remote tier: ~364 days (implicit: retention.ms - local.retention.ms)
Total retention: 1 year (retention.ms=31536000000)

Tiered Storage Configuration Reference

Configuration Default Description

Configuration	Default	Description
`remote.storage.enable`	`false`	Enable tiered storage for this topic (Kafka KIP-405)
`local.retention.ms`	`-2`	Time to keep in the local tier before Redis data expires. `-2` = use total `retention.ms`
`local.retention.bytes`	`-2`	Size to keep in the local tier before Redis trimming falls back to `retention.bytes`. `-2` = use total `retention.bytes`
`retention.ms`	`604800000`	Total retention across all tiers (7 days default)

remote.storage.enable

false

Enable tiered storage for this topic (Kafka KIP-405)

local.retention.ms

-2

Time to keep in the local tier before Redis data expires. -2 = use total retention.ms

local.retention.bytes

-2

Size to keep in the local tier before Redis trimming falls back to retention.bytes. -2 = use total retention.bytes

retention.ms

604800000

Total retention across all tiers (7 days default)

Remote tier retention is implicit and calculated as retention.ms - local.retention.ms. Data is deleted after the total retention.ms period.

See Remote Storage for server-level tiered storage configuration.