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 topics`

korvet 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 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 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 (default none). Use zstd for JSON/log storage efficiency, snappy when CPU cost matters more, and none when non-Kafka clients must read values directly with 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 none). This is distinct from the Kafka-facing compression.type, which only affects fetch-response compression. Prefer zstd for JSON/log topics and snappy for CPU-sensitive topics.

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 topics --alter or kafka-configs to alter topic configurations.

korvet 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 topics --describe or kafka-configs --describe:

korvet 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 none). It is independent of the Kafka-facing compression.type, which only affects fetch-response compression. Prefer zstd for JSON/log topics, snappy for CPU-sensitive topics, and none when non-Kafka clients must read values directly from Redis.

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.