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

Admin API

Korvet exposes an HTTP Admin API under /api/v1 for topic administration, consumer-group inspection, monitoring snapshots, broker security posture, and Kafka SASL credential management.

Authentication

REST API endpoints under /api/v1/ use JWT cookie authentication when korvet.admin.security-enabled=true (the default). POST /api/v1/auth/login validates an admin username and password, then sets an HttpOnly session cookie that the browser sends on later Admin API calls. HTTP Basic credentials are not accepted by the /api/v1/ REST API chain.

GET /api/v1/auth/status is public and reports whether first-run setup is required; when the request carries a valid session cookie it also reports the caller’s username and role. On a fresh secured deployment, create the first admin user with POST /api/v1/setup/admin while firstRunSetupRequired=true.

Every console user has a role: ADMIN (full read-write, the default) or VIEWER (read-only — GET succeeds, mutations return 403 Forbidden). POST /api/v1/users accepts an optional role, and PUT /api/v1/users/{username} changes the password, the role, or both. The last remaining ADMIN cannot be deleted or demoted (422 Unprocessable Entity). See Authentication for details.

When korvet.admin.security-enabled=false, all Admin API routes are public. Use that mode only for local single-user contexts such as korvet demo.

Swagger UI and the OpenAPI JSON endpoints are protected separately with HTTP Basic authentication so direct browser navigation can use the browser’s native login dialog. Actuator endpoints other than health probes also use HTTP Basic authentication — see Monitoring for details.

Login

POST /api/v1/auth/login
Content-Type: application/json

{
  "username": "admin",
  "password": "secret123"
}

The login endpoint returns a session cookie valid for 8 hours (configurable via korvet.admin.jwt.expiry).

Login attempts are rate-limited per username and client IP, and POST /api/v1/auth/logout revokes the session token immediately — see Authentication for the throttling policy and session-revocation details.

First-run setup

POST /api/v1/setup/admin
Content-Type: application/json

{
  "username": "admin",
  "password": "secret123"
}

Broker Security Status

  • GET /api/v1/security/status returns read-only TLS and SASL posture for the configured broker listener.

The response is metadata-only: it includes configured listener, TLS, and SASL flags, and never includes private keys, passwords, certificate contents, or local certificate file paths.

Console User Management

The Admin API manages the console users who access the web UI and REST API:

  • GET /api/v1/users — list console users

  • GET /api/v1/users/{username} — get details for one user

  • POST /api/v1/users — create a console user, with an optional role (ADMIN by default)

  • PUT /api/v1/users/{username} — change the password, the role, or both

  • DELETE /api/v1/users/{username} — delete a console user

The last remaining ADMIN cannot be deleted or demoted (422 Unprocessable Entity). See Authentication for the role model.

Password Change Security

When changing your own password, you must prove you know the current one:

PUT /api/v1/users/alice
Content-Type: application/json

{
  "currentPassword": "old-password",
  "password": "new-secure-password"
}

currentPassword is required when changing your own password: omitting it returns HTTP 400 Bad Request, and an incorrect value returns HTTP 403 Forbidden.

An admin resetting another user’s password must omit currentPassword (sending it returns an error); this lets admins reset accounts without knowing the current password while keeping the two flows distinguishable. The distinction prevents an attacker with a hijacked session from silently taking over an account by changing the password without proving they know the current one.

Kafka SASL Credentials

The credentials API manages Kafka SASL service accounts — the usernames and passwords that Kafka clients use to authenticate against the broker.

Supported mechanisms

  • SCRAM-SHA-256 (default) — recommended for most deployments.

  • PLAIN — requires TLS to be enabled on the broker (korvet.broker.tls=true).

Endpoints

  • POST /api/v1/credentials — create a credential.

  • GET /api/v1/credentials — list all credentials.

  • GET /api/v1/credentials/{username} — get one credential.

  • PUT /api/v1/credentials/{username} — rotate the password or change the mechanism.

  • DELETE /api/v1/credentials/{username} — delete a credential.

Create a credential

POST /api/v1/credentials
Content-Type: application/json

{
  "username": "kafka-client-1",
  "password": "secret123",
  "mechanism": "SCRAM-SHA-256"
}

mechanism is optional and defaults to SCRAM-SHA-256. username must be 3–64 characters and contain only letters, digits, -, _, or .. password must be 8–128 characters with no whitespace.

Rotate a password

PUT /api/v1/credentials/kafka-client-1
Content-Type: application/json

{ "password": "newSecret456" }

Omitting mechanism keeps the credential’s existing mechanism.

ACLs

/api/v1/acls manages the broker’s allow-only authorization rules. A rule grants a principal an operation (READ/WRITE) on a resource: resourceType is TOPIC (default) or GROUP (a consumer group id), and resource matches exact, prefix*, or *. The legacy {principal, topic, operation} request shape still works and creates a TOPIC rule; responses always carry resourceType and resource, plus topic for topic rules.

A principal is a SASL username or, with federated authentication, an IdP group written as group:<name> (for example group:payments-team) — the broker grants such a rule to every OAUTHBEARER connection whose configured groups claim carries <name>. The bare prefix (group: with no name) is rejected, and local credential/console usernames can never start with group: (their name pattern excludes :), so group principals cannot collide with users.

POST /api/v1/acls
Content-Type: application/json

{
  "principal": "alice",
  "resourceType": "GROUP",
  "resource": "billing-consumers",
  "operation": "READ"
}

  • GET /api/v1/acls — principals that have rules

  • GET /api/v1/acls/{principal} — a principal’s rules

  • DELETE /api/v1/acls/{principal}?resourceType=&resource=&operation= — delete one rule (the legacy topic parameter is still accepted for topic rules)

See Authentication for the matching semantics, the enforcement points, and the storage-format migration note.

Topics

  • POST /api/v1/topics creates a topic.

  • GET /api/v1/topics lists topics.

  • GET /api/v1/topics/{name} returns one topic.

  • GET /api/v1/topics/{name}/stats returns current per-topic traffic counters.

  • GET /api/v1/topics/{name}/partitions returns current per-partition stream and offset stats. Each row includes sizeBytes — the measured local Redis byte size (MEMORY USAGE) summed over the physical streams currently backing the partition (the logical stream, or every segment with a local copy). It is null when any component measurement is unavailable and 0 when the partition holds no local data (for example, fully offloaded). The same field appears on GET /api/v1/topic-partitions.

  • GET /api/v1/topics/{name}/messages browses stored topic messages without mutating consumer state.

  • PUT /api/v1/topics/{name} replaces explicit topic configuration.

  • DELETE /api/v1/topics/{name} deletes a topic.

Topic stats

  • GET /api/v1/topics/{name}/stats returns accumulated topic totals plus current-process traffic counters.

The response includes root topic, accumulated (producedMessages, producedBytes, nullable messageCount), and current (produceRecords, produceRequests, fetchRequests). current counters are process-local Micrometer values: they reset on restart, omit other brokers, and fetchRequests includes empty polls. Persisted consume totals, last activity timestamps, auto-created status, and throughput history require core tracking in a later phase.

Browse topic messages

The message browser endpoint is read-only. It uses existing Korvet storage reads and does not create consumer groups, commit offsets, reset offsets, or advance broker read state.

GET /api/v1/topics/orders/messages?partition=0&seek=earliest&limit=25&decode=utf8
Table 1. Query parameters
Parameter Description

partition

Optional partition number. Omit it to browse across all partitions.

seek

Read cursor. Supported values are earliest, offset, timestamp, and latest. latest is only valid together with direction=backward.

direction

Browse direction. forward (the default) reads from the seek position toward newer records; backward reads toward older records, returning the newest matching records first. seek=latest requires direction=backward, and seek=earliest with direction=backward browses the whole topic newest-first.

seekOffset

Required when seek=offset. Kafka offset in the selected partition or partitions.

timestamp

Required when seek=timestamp. Epoch milliseconds converted to a Redis Stream cursor. This matches the response row timestamp, which is the millisecond component of streamId.

offset

Zero-based page offset. Defaults to 0. offset + limit must not exceed 10000.

limit

Maximum rows to return. Defaults to 100; maximum is 100.

decode

Payload decode mode: raw for base64, utf8 for strings, json for best-effort JSON parsing, or schema for schema-registry decoding of Confluent wire-format payloads (Avro, Protobuf, or JSON Schema) through the embedded schema registry. Defaults to raw.

 
{
  "items": [
    {
      "topic": "orders",
      "partition": 0,
      "offset": 42,
      "streamId": "1782260333186-0",
      "timestamp": 1782260333186,
      "recordTimestamp": 1782260333000,
      "key": {
        "preview": "order-42",
        "value": "order-42",
        "encoding": "utf8",
        "size": 8,
        "truncated": false,
        "error": null,
        "decodeStatus": "decoded"
      },
      "value": {
        "preview": "{\"id\":42}",
        "value": "{\"id\":42}",
        "encoding": "utf8",
        "size": 9,
        "truncated": false,
        "error": null,
        "decodeStatus": "decoded"
      },
      "headers": [],
      "size": 32,
      "tier": null
    }
  ],
  "total": 1,
  "offset": 0,
  "limit": 25,
  "nextOffset": null
}

headers lists the record’s headers in stored order (duplicate keys preserved); each entry has a key and a value field decoded like key/value above. It is [] when the record has no headers and null when the stored header blob cannot be decoded. timestamp is the stream-id timestamp used by seek=timestamp; recordTimestamp is the stored Kafka record timestamp and may be null. offset and limit follow the same page envelope as GET /api/v1/storage/offload-jobs; total is the number of visible messages addressable by the bounded offset paginator from the selected seek cursor before the page offset is applied. It is capped by the offset + limit maximum page window. nextOffset is the page offset of the next page, or null when the result set is exhausted. tier is always null.

Each decoded field (key, value, and header values) carries a decodeStatus:

  • decoded — the payload was decoded with its registered schema.

  • raw — the payload was rendered as-is (no schema decoding attempted).

  • schema-not-found — the record’s Confluent wire-format schema id was never registered; the field falls back to raw base64 rendering with the reason in error. Deleting a schema subject or version keeps its schema ids resolvable, so records written before the delete still decode.

  • decode-error — the schema was found but decoding failed; the field falls back to raw base64 rendering with the reason in error.

A decode failure affects only that field: the page never fails because a single record could not be decoded.

Consumer Groups

  • GET /api/v1/consumer-groups lists consumer groups.

  • GET /api/v1/consumer-groups/{groupId} returns group state and members.

  • GET /api/v1/consumer-groups/{groupId}/offsets returns current per-partition offsets, end offsets, undelivered-message lag, and available stream group metadata. Optional filters are topic and partition. Lag is storage-reported message count and excludes delivered-but-uncommitted pending messages.

  • POST /api/v1/consumer-groups/{groupId}/reset-offsets clears the group’s committed offsets and stream delivery state; consumers restart according to their own auto.offset.reset policy. Responds 204. With ?dryRun=true nothing is mutated: the response (200) previews the per-partition state the reset would clear — groupId, partitionCount, and partitions rows with topic, partition, committedOffset, currentOffset, endOffset, pendingCount, and targetOffset. Offsets are serialized as strings; targetOffset is always null because the reset clears offset state rather than committing a broker-chosen offset. Only partitions that carry group state are listed. Requires the ADMIN role.

Configuration

  • GET /api/v1/configuration/runtime-policies returns read-only runtime policy values from the running broker configuration.

currentValue and defaultValue are raw config strings: durations use milliseconds, sizes use bytes, and enums use their Kafka/config names. User interfaces should format these values for display.

  • GET /api/v1/configuration/system returns read-only, effective startup-bound system configuration for the running broker/API process.

Values are machine-readable: durations use Ms fields, sizes use bytes, and enum/config names are returned as strings. Secrets are omitted and exposed only as *Configured flags. When redis.uriConfigured=true, Redis host, port, and credential flags are derived from the URI.

Runtime Logging

  • GET /api/v1/loggers returns runtime logger levels for ROOT, shallow Korvet-owned loggers under com.redis.korvet.*.

  • POST /api/v1/loggers/{name} sets or clears one visible logger’s runtime level.

The response includes supported levels, items with name, nullable configuredLevel, and effectiveLevel, and total. Send configuredLevel=null to clear an explicit runtime level. Runtime log-level changes are process-local and do not survive restart unless also configured at startup.

Monitoring

  • GET /api/v1/health returns Spring Boot health under details plus backend-owned checks for the Operations UI. The details tree includes redis (broker Redis reachability, plus the local-storage Redis backend when configured separately) and broker (local broker liveness) subsystems, so status degrades to DOWN when either backing store is unreachable.

  • GET /api/v1/metrics returns a curated metrics snapshot: topic/group counts, current activeConnections and backpressureConnections gauges, cumulative messagesIn/bytesIn/bytesOut/produceRequests/ fetchRequests/rebalances/brokerFailures/lossyRecords counters, and requestLatency percentiles aggregated across all Kafka API request timers.

  • GET /api/v1/metrics/topics returns per-topic produce/fetch rates (records/s, bytes/s, requests/s) and produce/fetch latency percentiles (p50Ms/p95Ms/p99Ms), paginated by topic name via offset/limit (default 100, max 1000) with total for detecting partial pages, and optionally filtered to a single topic via topic. Rates are per-step counter deltas from the in-process metrics history store averaged over a rolling lookback window, so they read 0 until the first step (korvet.admin.metrics-history.interval, default 15s) completes; percentile fields are null until the topic has seen traffic. Each entry also carries samples — the topic’s per-step produced-traffic history (timestamp, produceRecordsPerSecond, produceBytesPerSecond) over the retention window, oldest first, backing the per-topic trend sparklines. The array is empty when the topic has no history series, e.g. it was rejected by the korvet.admin.metrics-history.max-series cap or the broker just restarted.

  • GET /api/v1/metrics/history returns curated per-step metrics captured in-process (defaults: every 15s with 1h retention, configurable via korvet.admin.metrics-history.*), oldest first. Counter-backed fields are per-second rates over each step, so clients never do delta math. The buffer is in-memory only — it starts empty on every restart and holds at most retention / interval samples — and backs the dashboard sparklines.

  • GET /api/v1/metrics/consumer-lag returns per-group aggregate lag (totalLag, maxLag, partitionsWithUnknownLag, lastCommitTimestamp) in one call, sized for dashboard polling. Totals are lower bounds when some partitions have unknowable lag, and null when none is known.

  • GET /api/v1/brokers returns live broker registry nodes enriched with the runtime metadata each broker publishes with its registry heartbeat: bindHost, bindPort, tls, sasl, saslMechanism, connections, lastHeartbeat (ISO-8601 instant), uptime (ISO-8601 duration), and version. Every enrichment field is nullable and degrades independently — entries written by brokers on builds that predate per-node metadata carry none of them, in which case the local node falls back to its own listener posture. A node is stale once lastHeartbeat is older than the registry’s 30-second liveness threshold (stale nodes are not listed).

  • GET /api/v1/storage-stats returns storage diagnostics: queueDepth, segmentsPerMin/bytesPerMin throughput, archiveLatency percentiles, cumulative archiveFailures by error type, and lagOldestSeconds — the age of the oldest un-archived segment (null until the storage worker reports lag, 0 when the backlog is empty).

Health checks contain stable id, name, status, severity, scope, targetLabel, message, nullable durationMs, lastCheckedAt, and actions. Checks cover Admin API response, the Actuator aggregate, broker liveness, broker registry, broker Redis connectivity, remote storage diagnostics, and schema-registry availability when the corresponding runtime beans are present. When local stream storage is configured with a separate Redis backend, an additional storage-redis-connectivity check identifies that target separately. Unsupported actions are reported as an empty actions array.

Offload Jobs

  • GET /api/v1/storage/offload-jobs lists segment offload jobs, merged from live tiered segment state and the durable job records the storage worker maintains. Query parameters: status (repeatable; pending, running, done, failed, cancelled), offset (zero-based row offset, default 0), and limit (default 100, maximum 500). Each item carries id, topic, partition, segment, status, attempts, maxAttempts, and — once the worker has attempted the job — startedAt, finishedAt, durationMs, and error (null otherwise). The counts object tallies every status over the whole population, independent of filter and pagination.

  • POST /api/v1/storage/offload-jobs/{id}/retry resets a failed or cancelled job to pending with zero attempts so the storage worker’s next offload scan retries the segment. Returns the updated job; responds 409 for any other status and 404 for unknown jobs.

  • POST /api/v1/storage/offload-jobs/{id}/cancel cancels a pending or failed job: the worker skips the segment until the job is retried. The segment’s local data is kept (and, without an offloaded copy, is not freed by local retention). Returns the updated job; responds 409 for running, done, or already-cancelled jobs.

Job lifecycle: a sealed segment surfaces as pending; each worker tick starts an attempt (running) that ends done or failed. Failed jobs are retried automatically every tick until attempts reaches maxAttempts (korvet.storage.worker.offload-jobs.max-attempts, default 5), after which the job stays failed until retried manually. Finished (done) records are kept up to korvet.storage.worker.offload-jobs.retention-count (default 1000); failed and cancelled records are exempt from that cap. Records for segments already deleted by retention remain visible as historical rows until evicted. Both POST endpoints require the ADMIN role; VIEWER sessions can only read.

Redis Streams Inspection

  • GET /api/v1/storage/redis-streams returns point-in-time metadata for the physical Redis Streams that currently back configured topic partitions. Each entry carries topic, partition, the physical stream key, length, firstId, lastId, memoryBytes, and a status of active, empty, or unavailable. The inventory is derived from topic and segment metadata and does not scan Redis, so missing streams stay distinguishable from unavailable metadata reads.

  • GET /api/v1/storage/redis-streams/runtime returns the local write circuit-breaker runtime state as a circuitBreaker object with state (disabled, closed, open, or half-open) and, only while open, openUntil. Failure details are not exposed.

memoryBytes is the stream’s measured physical Redis memory (MEMORY USAGE) and is null when the measurement is unavailable. The circuit breaker is the write-path breaker described in Production Tuning; while it is open, produces fail fast with the retriable REQUEST_TIMED_OUT error.

Storage Segments

  • GET /api/v1/topics/{name}/partitions/{partition}/segments/{segmentId} returns detailed information about a specific segment, including local Redis metadata and remote Iceberg metadata when the segment is offloaded.

Segment Detail Response

For offloaded segments with tiered storage enabled, the response includes a remoteStore object describing the segment’s Iceberg data files:

{
  "topic": "orders",
  "partition": 0,
  "segmentId": 1,
  "state": "OFFLOADED",
  "tier": "remote",
  "open": false,
  "messageCount": 10000,
  "byteSize": 524288,
  "startId": "1722153600000-0",
  "endId": "1722153900000-42",
  "startTimestamp": 1722153600000,
  "endTimestamp": 1722153900000,
  "localStore": {
    "streamKey": "korvet:storage:local:orders:0:1",
    "length": 10000,
    "firstId": "1722153600000-0",
    "lastId": "1722153900000-42"
  },
  "remoteStore": {
    "format": "PARQUET",
    "table": "analytics.orders",
    "catalog": "rest",
    "dataFiles": [
      {
        "location": "s3://my-bucket/korvet/analytics/orders/data/00000-0.parquet",
        "bucket": "my-bucket",
        "objectKey": "korvet/analytics/orders/data/00000-0.parquet",
        "sizeBytes": 524288,
        "recordCount": 10000
      }
    ]
  }
}

The remoteStore object includes:

  • format: data file format (PARQUET)

  • table: the Iceberg table name

  • catalog: Iceberg catalog type (e.g., rest)

  • dataFiles: the segment’s data files, each with location, bucket, objectKey, sizeBytes, and recordCount

The remoteStore field is:

  • Present with metadata when the segment is offloaded and remote inspection succeeds

  • null when the segment is not offloaded, remote inspection fails, or times out (not an error — the response still includes complete local metadata)

Similarly, localStore carries the local Redis stream metadata (streamKey, length, firstId, lastId) and is null once a segment exists only on the remote tier.

See Remote Segment Metadata Inspection for inspection behavior, caching, and configuration.

OpenAPI

OpenAPI JSON is available at /v3/api-docs.

Swagger UI is available at /swagger-ui.html.