spoud/kapture repository overview

⁠Kapture

Capture Kafka topics into DuckDB and query them with SQL.

Kapture runs as a single Quarkus service. It subscribes to one or more Kafka topics, applies a configurable SMT chain, writes records into DuckDB, and exposes a SQL query endpoint plus shared saved-query storage with a Monaco Editor web UI.

Kafka topics ──► KafkaConsumerService ──► SMT chain ──► DuckDbWriter ──► kapture.duckdb
                                                                                 │
                                                                        DuckDbQueryService
                                                                                 │
                                                                     POST /api/query
                                                                                  │
                                                                     GET/POST /api/saved-queries

⁠Quick start (Docker Compose)

Starts Redpanda, produces sample fixtures, and runs Kapture against them:

docker compose -f deploy/docker-compose.yml up --build

Open http://localhost:8080⁠ in a browser. The Monaco Editor UI lists synced topics in the sidebar, lets you run SQL against them (Ctrl+Enter or the Run button), and can store named shared queries on the server for later reuse by any user.

The bundled demo also includes a customers_masked topic that uses both fixed masking and deterministic hashing before rows are written to DuckDB: credit-card numbers and passwords are replaced with ***masked***, while email and socialSecurityNumber are pseudonymized with a keyed hash so the same input stays joinable across rows without exposing the raw value.

SELECT customerId, email, creditCardNumber, password, socialSecurityNumber
FROM customers_masked
ORDER BY customerId
LIMIT 5

You can inspect the exact SMT inventory of the running demo with:

curl http://localhost:8080/api/smts

Stop and clean up:

make docker-down

⁠UI preview

The gallery below comes from the bundled Docker Compose demo stack. The screenshots are linked from the README so GitHub and DockerHub can both render them.

⁠Build & test

# Build (skip tests)
make build

# Unit tests for SMT library — fast, no Docker required
make test-smts

# Integration tests — requires Docker (Testcontainers + Redpanda)
make test-integration

# All tests
make test

# Quarkus dev mode with live reload
# Set KAFKA_BOOTSTRAP_SERVERS and KAPTURE_TOPICS first (or edit application.properties)
make dev

⁠Versions

⁠Configuration

All configuration is driven by environment variables (or application.properties overrides).

⁠Required

⁠Kafka authentication

The default security profile is Confluent Cloud (SASL_SSL + PLAIN). All auth properties are optional and override-able.

Confluent Cloud (SASL_SSL + PLAIN API key/secret):

KAFKA_SECURITY_PROTOCOL=SASL_SSL
KAFKA_SASL_MECHANISM=PLAIN
KAFKA_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="<API_KEY>" password="<API_SECRET>";

Self-hosted cluster (SCRAM-SHA-512):

KAFKA_SECURITY_PROTOCOL=SASL_SSL
KAFKA_SASL_MECHANISM=SCRAM-SHA-512
KAFKA_SASL_JAAS_CONFIG=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="<user>" password="<pass>";

mTLS (add on top of any of the above):

KAFKA_SSL_TRUSTSTORE_LOCATION=/certs/truststore.jks
KAFKA_SSL_TRUSTSTORE_PASSWORD=<pw>

No auth (local/dev):

KAFKA_SECURITY_PROTOCOL=PLAINTEXT

⁠General options

⁠Topic reset API

Kapture exposes an operator endpoint to rebuild one configured topic from Kafka from the beginning:

curl -X POST http://localhost:8080/api/topics/<topic>/reset

The reset drops the topic table and its parsing-issues table from DuckDB, rewinds the current consumer group for that topic to the earliest retained offsets, and resumes ingestion. This is useful after changing per-topic config such as raw-column storage, SMTs, or converters.

To restrict this endpoint to a narrower operator role, add a more specific permission for the reset path ahead of the catch-all /* rule and assign that role only to the users that should be allowed to rewind topics:

quarkus.http.auth.permission.public.paths=/q/health,/q/health/*,/q/metrics
quarkus.http.auth.permission.public.policy=permit

quarkus.http.auth.permission.topic-reset.paths=/api/topics/*/reset
quarkus.http.auth.permission.topic-reset.policy=topic-reset-policy
quarkus.http.auth.policy.topic-reset-policy.roles-allowed=KaptureAdmin

quarkus.http.auth.permission.kapture-role.paths=/*
quarkus.http.auth.permission.kapture-role.policy=role-policy
quarkus.http.auth.policy.role-policy.roles-allowed=KaptureUser

With basic auth, assign KaptureAdmin only to the selected operators. With OIDC, map that role from the identity provider and keep regular users on KaptureUser.

⁠Installed SMT catalog

Kapture exposes the exact SMTs bundled in the running image at:

curl http://localhost:8080/api/smts

The response includes the class name, provider (kapture or apache-kafka), scope (record, key, or value), and the required/optional config keys for each installed SMT. This endpoint is the authoritative source for what your running image can load.

⁠Per-topic config

Each topic gets a converter and an optional SMT chain.

Topic names with dots work as-is in application.properties. For env vars, dots in topic names map to underscores (e.g. topic com.example.orders → KAPTURE_TOPICS_COM_EXAMPLE_ORDERS_CONVERTER).

# Converter: JsonConverter (default), JsonSchemaConverter, StringConverter, AvroConverter, or ProtobufConverter
kapture.topics.<topic>.converter=JsonConverter

# JsonSchemaConverter / AvroConverter / ProtobufConverter: point at your Schema Registry (global default or per-topic override)
kapture.schema-registry.url=http://localhost:8081
kapture.schema-registry.username=<username>                      # optional
kapture.schema-registry.password=<password>                      # optional
kapture.topics.<topic>.schema-registry-url=http://other-registry:8081   # optional override
kapture.topics.<topic>.schema-registry-username=<username>       # optional override
kapture.topics.<topic>.schema-registry-password=<password>       # optional override

# Optional SMT chain (comma-separated names)
kapture.topics.<topic>.transforms=t1,t2
kapture.topics.<topic>.transforms.t1.type=io.spoud.kapture.smts.TombstoneFilter
kapture.topics.<topic>.transforms.t2.type=io.spoud.kapture.smts.ExtractArrayFirstElement
kapture.topics.<topic>.transforms.t2.field=LedgerEntry

# Example using a fixed mask plus deterministic keyed hashing
kapture.topics.customers_masked.transforms=maskpii,hashpii
kapture.topics.customers_masked.transforms.maskpii.type=org.apache.kafka.connect.transforms.MaskField$Value
kapture.topics.customers_masked.transforms.maskpii.fields=creditCardNumber,password
kapture.topics.customers_masked.transforms.maskpii.replacement=***masked***
kapture.topics.customers_masked.transforms.hashpii.type=io.spoud.kapture.smts.HashField
kapture.topics.customers_masked.transforms.hashpii.fields=email,socialSecurityNumber
kapture.topics.customers_masked.transforms.hashpii.secret=${KAPTURE_HASH_SECRET}
kapture.topics.customers_masked.transforms.hashpii.length=16

# Optional ART index on the key column (speeds up key-based WHERE / tombstone UPDATEs)
kapture.topics.<topic>.index-key=true

# Optional per-topic flush threshold for very large records
kapture.topics.<topic>.flush-batch-size=100

# Optional cleanup when raw storage has been disabled for an existing table
kapture.topics.<topic>.drop-raw-columns-when-disabled=true

# Optional raw-column storage (defaults to true for all three)
kapture.topics.<topic>.store-raw-key=false
kapture.topics.<topic>.store-raw-value=false
kapture.topics.<topic>.store-raw-headers=false

If you configure SMT classes through shell environment variables, escape the inner-class separator in MaskField$Value so your shell does not drop the $Value suffix. Examples:

export KAPTURE_TOPICS_CUSTOMERS_MASKED_TRANSFORMS_MASKPII_TYPE='org.apache.kafka.connect.transforms.MaskField$Value'
# or: org.apache.kafka.connect.transforms.MaskField\$Value

In application.properties and Kubernetes YAML values, the plain MaskField$Value form is correct. In Docker Compose YAML, write MaskField$$Value so Compose does not treat $Value as variable interpolation.

KAPTURE_KAFKA_MAX_POLL_RECORDS is global because Kafka applies it per consumer, not per topic. If only some topics are unusually large, keep the poll limit moderate and set kapture.topics.<topic>.flush-batch-size lower for those specific topics so Kapture writes them out sooner.

JsonSchemaConverter:

Deserialises Confluent Schema Registry wire-format JSON Schema messages (magic byte + schema ID + JSON payload). The resulting payload is normalised to plain Java Map/List/primitive values, so FlattenJson and DuckDB writes work exactly like for plain JSON topics.

# Example: JSON Schema topic with nested JSON flattened into columns
kapture.schema-registry.url=http://redpanda:8081
kapture.topics.products.converter=JsonSchemaConverter
kapture.topics.products.transforms=flattedjson
kapture.topics.products.transforms.flattedjson.type=io.spoud.kapture.smts.FlattenJson

AvroConverter:

Deserialises Confluent Schema Registry wire-format Avro messages. The schema is fetched automatically from the registry on first use and cached. The resulting GenericRecord is converted to a flat Map<String, Object> identical in shape to a JsonConverter result.

Avro type mapping: string/Utf8 → String, int/long/float/double/boolean → boxed primitive, enum → String, bytes/fixed → Base64 string, nested record/array/map → serialised to a JSON string (use FlattenJson SMT to promote nested fields into columns).

# Example: Avro topic with nested record flattened into columns
kapture.schema-registry.url=http://redpanda:8081
kapture.topics.transactions.converter=AvroConverter
kapture.topics.transactions.transforms=flattedjson
kapture.topics.transactions.transforms.flattedjson.type=io.spoud.kapture.smts.FlattenJson
kapture.topics.transactions.index-key=true

ProtobufConverter:

Deserialises Confluent Schema Registry wire-format Protobuf messages and converts them to plain JSON-compatible maps before they are flattened and written to DuckDB.

kapture.topics.inventory.converter=ProtobufConverter
kapture.schema-registry.url=http://schema-registry:8081

For schema-registry-backed converters, Kapture injects these internal metadata fields during ingestion when available:

• _schema_id_key
• _schema_id_value
• _schema_header
• _schema_payload

Raw system columns are written as _raw_key, _raw_value, and _raw_headers. They can be disabled per topic with the store-raw-* flags above when you want to avoid duplicating the original payload bytes in DuckDB. If you rely on tombstone deletes or key-based indexing, keep _raw_key enabled for that topic.

SMTs:

Any class implementing the Kafka Connect Transformation interface can be used. Kapture already ships with Kafka's connect-transforms library on the classpath, so the commonly used built-ins documented in Confluent guides (for example MaskField, ReplaceField, or TimestampConverter) work without adding extra jars. Classes are loaded by fully-qualified name; use /api/smts to see the exact runtime inventory.

In the Docker image, the bundled SMT classes are included at build time through the normal Maven dependencies:

• Kafka/Connect SMTs such as org.apache.kafka.connect.transforms.MaskField$Value come from org.apache.kafka:connect-transforms
• Kapture-specific SMTs come from the sibling kapture-smts module
• Quarkus packages them into the fast-jar layout under /app/lib/ and /app/app/, and the container starts with java -jar /app/quarkus-run.jar

Kapture now also supports mounted external SMT jars at runtime. By default it looks in /app/ext (override with KAPTURE_SMTS_EXTRA_DIR) and loads any classes from mounted jars that implement Kafka Connect's Transformation interface.

Mounting additional SMT jars

docker run \
  -v /path/to/my-smts:/app/ext:ro \
  -e KAPTURE_SMTS_EXTRA_DIR=/app/ext \
  -e KAPTURE_TOPICS_ORDERS_TRANSFORMS=maskpii \
  -e KAPTURE_TOPICS_ORDERS_TRANSFORMS_MASKPII_TYPE=com.example.kafka.MaskSensitiveFields \
  spoud/kapture:latest

For Docker Compose:

services:
  kapture:
    volumes:
      - ./smts:/app/ext:ro
    environment:
      KAPTURE_SMTS_EXTRA_DIR: /app/ext
      KAPTURE_TOPICS_ORDERS_TRANSFORMS: maskpii
      KAPTURE_TOPICS_ORDERS_TRANSFORMS_MASKPII_TYPE: com.example.kafka.MaskSensitiveFields

If an external SMT jar has additional dependencies, mount those dependency jars into the same directory as well so the external SMT classloader can resolve them. Mounted external SMTs appear in /api/smts with provider: external.

Common Kafka built-in (org.apache.kafka.connect.transforms.*) examples:

Kapture-specific (io.spoud.kapture.smts.*):

ExplodeArray — one row per array element:

Topics where a single Kafka message carries a list of items (e.g. a purchase order with multiple line items) can be exploded so that each item becomes its own DuckDB row. This makes every item field directly queryable without JSON string parsing.

# Each purchase order message explodes into one row per line item,
# then FlattenJson promotes every item field to a top-level column.
kapture.topics.purchase_orders.transforms=explodeitems,flattenjson
kapture.topics.purchase_orders.transforms.explodeitems.type=io.spoud.kapture.smts.ExplodeArray
kapture.topics.purchase_orders.transforms.explodeitems.field=order.items
kapture.topics.purchase_orders.transforms.flattenjson.type=io.spoud.kapture.smts.FlattenJson

For topics that use ExplodeArray, Kapture automatically extends the primary key to (partition, offset, _explode_index) so that all rows from the same Kafka message can coexist. _explode_index is a 0-based integer injected by the SMT and treated as an internal system column (hidden by default in the UI). When store-raw-value / store-raw-key / store-raw-headers is enabled, raw columns are populated only on the first row per message (_explode_index = 0) to avoid duplicating the full original payload across every exploded row.

ExtractHeader — lifting Kafka headers into columns:

A common pattern is attaching tracing context (W3C traceparent, Zipkin X-B3-TraceId) or routing metadata as Kafka headers. ExtractHeader promotes a named header to a queryable DuckDB column.

# Extract the W3C traceparent header into a trace_id column, then flatten nested JSON
kapture.topics.orders.transforms=extracttrace,flattedjson
kapture.topics.orders.transforms.extracttrace.type=io.spoud.kapture.smts.ExtractHeader
kapture.topics.orders.transforms.extracttrace.header=traceparent
kapture.topics.orders.transforms.extracttrace.field=trace_id
kapture.topics.orders.transforms.flattedjson.type=io.spoud.kapture.smts.FlattenJson

The transactions topic in the bundled demo uses this pattern — producers set a traceparent header and the SMT chain promotes it to a trace_id column, allowing queries like:

-- All transactions that belong to the same distributed trace
SELECT id, account, amount, type
FROM transactions
WHERE trace_id = '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01';

-- Aggregate spend per trace (e.g. to correlate with slow APM traces)
SELECT trace_id, count(*) AS txn_count, sum(amount) AS total_amount
FROM transactions
GROUP BY trace_id
ORDER BY total_amount DESC;

If a record has no traceparent header the trace_id column is NULL — absent headers never cause errors, which handles mixed producers (some instrumented, some not) gracefully.

Tombstone / compacted-topic semantics:

When a Kafka record has a null value (tombstone), Kapture can react in one of two ways, controlled per topic by tombstone-mode:

No SMT is required for either mode — tombstones are intercepted before the SMT chain and routed through the JDBC ordered path, interleaved correctly with adjacent INSERTs.

Soft-delete configuration:

kapture.topics.<topic>.tombstone-mode=soft-delete

Or via env var (dots → underscores):

KAPTURE_TOPICS_PRODUCTS_TOMBSTONE_MODE=soft-delete

When soft-delete is active, two columns are added automatically to the topic's table if they don't already exist:

Example queries:

-- Active records only
SELECT * FROM products WHERE _deleted = false;

-- All soft-deleted records and when they were removed
SELECT key, _deleted_at FROM products WHERE _deleted = true ORDER BY _deleted_at;

-- Full history including deleted rows
SELECT * FROM products ORDER BY key, ts;

⁠Duplicate / re-delivery handling

Every topic table has a PRIMARY KEY (partition, offset). For topics using ExplodeArray, the key is extended to PRIMARY KEY (partition, offset, _explode_index) so all rows from the same Kafka message can coexist. In both cases, re-delivered records are silently dropped via INSERT … ON CONFLICT … DO NOTHING. Kafka offsets are committed only after a successful DuckDB flush, so at-least-once delivery is guaranteed and re-delivery is idempotent.

This covers Kafka-level duplicates (same partition + offset). Producer-level duplicates — where the same business event was published twice with two distinct offsets — are stored as separate rows. Deduplicate those at query time with QUALIFY ROW_NUMBER() OVER (PARTITION BY <business_key> ORDER BY ts DESC) = 1.

⁠DuckDB schema

Every topic gets its own table:

⁠DuckDB tuning

Query access mode (KAPTURE_QUERY_ACCESS_MODE):