Minimalist Kafka¶

Guide: the opt-in minimalist-kafka library — route Kafka topics into Event Script flows, publish events to Kafka, and health-check the cluster.

At a glance

What — minimalist-kafka is an opt-in library with two composable building blocks: an inbound Kafka Flow Adapter that routes each topic (or regex-matched set of topics) into an Event Script flow (the Kafka counterpart of rest.yaml), and an outbound notification function that publishes an event to a topic.

Config, not code — Kafka client connection/security comes from external kafka-producer.properties / kafka-consumer.properties templates; the YAML binds topics (literal or regex) to flows with optional consumer group, partition pinning, a per-binding dead-letter topic, and a per-binding delivery mode. Enterprise SASL/OAuth2/mTLS is configured, never coded.

Reliable, with a throughput escape hatch — at-least-once consume (commit-after-process) by default, bounded retry then a per-binding dead-letter topic, and continuous W3C trace context across the Kafka hop; a binding may opt into Kafka-native auto-commit for higher throughput instead.

For developers and operators triggering flows from Kafka, or emitting Kafka events from a flow.

The built-in HTTP flow adapter routes HTTP requests into flows. minimalist-kafka does the same for Kafka: a topic listener mints an EventEnvelope and hands it to the Event Script engine, so the flow's tasks — not the I/O layer — do the work. It is not the service mesh (cloud.connector=kafka), which is a different concern; this library is an application-level building block you opt into.

This is an opt-in library. Add the minimalist-kafka dependency and set yaml.kafka.flow.adapter to activate the inbound adapter. The outbound simple.kafka.notification function registers automatically.

Enabling the library¶

Depend on system/minimalist-kafka (it depends on event-script-engine):
```
<dependency>
    <groupId>org.platformlambda</groupId>
    <artifactId>minimalist-kafka</artifactId>
    <version>x.y.z</version>  
</dependency>
```
2. Point yaml.kafka.flow.adapter at your adapter config (inbound). Without it, no consumer starts. 3. Provide the Kafka client templates (see client config) — the classpath defaults work for local dev.

yaml.kafka.flow.adapter=classpath:/kafka-flow-adapter.yaml

The library autoloads at startup (@MainApplication): it builds the shared producer and, if yaml.kafka.flow.adapter is set, starts one consumer thread per topic binding. Either client can be switched off when the cluster has no credentials for it.

Switching off a client you do not use¶

Both clients start by default. When the cluster grants credentials for only one of them — the usual case for one leg of a bridge, where a Confluent console issues an API key for producing or consuming — switch the unused one off:

# a consume-only leg: this cluster issues no producer credentials
kafka.producer.enabled=false

Setting	Effect when `false`
`kafka.producer.enabled`	No producer is built. `simple.kafka.notification` stays registered but fails with a message naming this key, so a flow that publishes anyway points at the config rather than at a missing route.
`kafka.consumer.enabled`	No adapter consumer starts, even with `yaml.kafka.flow.adapter` set, and `kafka.health` probes through the producer template instead.

Two rules worth knowing:

The flag is a veto, not a trigger. Leaving it at the default starts nothing that is not otherwise configured — an inbound adapter still needs yaml.kafka.flow.adapter. Only the literal false switches a client off; any other value leaves it on.
A dead-letter topic needs a producer. Dead letters are published through this cluster's own producer, so a binding that declares dlq-topic while kafka.producer.enabled=false fails the deployment at startup, naming both settings. It is the contradiction that matters: without the guard an exhausted message would be dropped with a DATA LOSS log and its offset committed. Enable the producer, or drop the dlq-topic.

Disabling both is allowed — the module goes inert and says so with a startup WARN — which makes a "Kafka off in this profile" switch possible without removing the dependency.

Inbound: the adapter YAML¶

kafka-flow-adapter.yaml lists topic -> flow bindings:

consumer:
  - topic: 'incoming-orders'
    flow: 'process-order'
    group: 'sales-order-group'        # optional
    dlq-topic: 'incoming-orders-dlq'  # optional; no DLQ if omitted (failed messages dropped w/ ERROR)
  - topic: 'incoming-payments'
    flow: 'process-payment'
    partition: 0                      # optional
  - topic-pattern: 'events\.[a-z]{2}' # optional; regex subscribe instead of a literal 'topic'
    flow: 'process-region-event'
    group: 'region-events-group'      # required for topic-pattern bindings
  - topic: 'clickstream'
    flow: 'ingest-clickstream'
    auto-commit: true                 # optional; trades pod-death redelivery for throughput
    max-poll-records: 500             # optional; only meaningful with auto-commit
  - topic: 'mixed-events'             # second-level routing: pick the target per record
    serializer: 'json'                # optional; best-effort JSON decode on a non-schema topic
    flows:
      - 'input.header.type(order) -> flow://order-flow'
      - 'input.body.event.kind(refund) -> task://v1.refund.processor'
      - 'default -> flow://catch-all-flow'

Field	Required	Description
`topic`	one of `topic`/`topic-pattern`	Literal source Kafka topic.
`topic-pattern`	one of `topic`/`topic-pattern`	Regex subscription instead of a literal topic (see pattern subscription).
`flow`	one of `flow`/`flows`	Event Script flow id every message of this binding is routed into (direct routing).
`flows`	one of `flow`/`flows`	Second-level routing rule list — inspect a key-value of each record to pick the target flow or function per message (see second-level routing).
`group`	no (required for `topic-pattern`)	Consumer group id (see consumer group). Defaults to `kafka-flow-adapter.<topic>` for a literal topic; no default exists for a pattern.
`partition`	no	Pins a single partition (see partition pinning). Omit for group-managed assignment. Cannot be combined with `topic-pattern`.
`schema.enabled`	no	When `true`, decode the Confluent-framed value into a `Map` before routing it into the flow (see Schema Registry). Default `false` (raw `byte[]`).
`serializer`	no	`'json'` = best-effort SimpleMapper decode of the record value on a non-schema topic (see payload prerequisites). Mutually exclusive with `schema.enabled`.
`ttl`	no	Deadline for `task://` routing targets (duration syntax, e.g. `30s`, `5m`; default 30s) — a bare function has no flow ttl. Flow targets always use their own flow `ttl`.
`dlq-topic`	no	Pre-provisioned topic for exhausted messages (see reliability). No DLQ if omitted.
`auto-commit`	no	When `true`, use Kafka-native auto-commit instead of the default manual commit-after-process (see delivery mode). Default `false`.
`max-poll-records`	no	Override the delivery mode's default poll batch size (1 for manual-commit, 500 for auto-commit).
`correlation.id.header`	no	Per-binding override of the global `kafka.correlation.id.header` (default `cid`) — impedance matching for an upstream that publishes its own correlation-id header name (e.g. `X-Correlation-ID`).
`trace.id.header`	no	Per-binding override of the global `kafka.trace.id.header` — a fallback trace-id source for an upstream that does not send a W3C `traceparent` (which always takes precedence).
`traceparent.header`	no	Per-binding override of the global `kafka.traceparent.header` (default `traceparent`) — the header carrying the full W3C trace context, for backward compatibility with a legacy upstream only (departure from the W3C/OTel standard is discouraged). The standard `traceparent` always wins; the custom name is read only when the standard is absent.

The file is read by ConfigReader, so every value supports ${ENV_VAR:default} substitution — e.g. group: '${KAFKA_CONSUMER_GROUP:sales-order-group}'. A malformed entry (missing topic/topic-pattern, missing or duplicated flow/flows, a malformed routing rule or one referencing an unknown flow or task route, serializer combined with schema.enabled, an invalid regex, a dlq-topic that equals or matches its own source, etc.) fails startup fast and loud rather than being silently skipped.

Message dataset¶

Every message hands the flow a Map with three top-level objects — input.body, input.header, and input.metadata:

Field	Type	Description
`body`	`byte[]` or `Map`	The message payload; a `Map` when `schema.enabled` decodes a Confluent-framed value or `serializer: 'json'` parses a JSON object, raw `byte[]` otherwise.
`header`	`Map<String,String>`	The record's Kafka headers, including `traceparent` (consumed for trace continuity) and `cid` (correlation id) when the producer set them.
`metadata`	`Map<String,Object>`	The record's own envelope facts — `topic`, `partition`, `offset`, `timestamp` (epoch milliseconds, a `long`), and `key` (omitted when the record carries no key).

metadata.topic and metadata.partition are the record's actual topic and partition — not the binding's configured topic/topic-pattern. For a literal topic binding this is redundant (the flow already knows the topic from its own YAML), but for a topic-pattern binding it's the only way a flow recovers which of the many matched topics a given message came from, since every matched topic shares one flow. It's equally useful for a reprocessing flow bound to a dlq-topic: metadata.topic there is the DLQ topic itself, while the dlq.origin.topic header (see reliability) carries the original source topic — together they let a reprocessor recover both "where this landed" and "where it came from" without any framework-side rule/schema code.

Because metadata is just another field on input, a task's own input: mapping can pass it straight to a composable function's parameter — no model.* relay needed. This is what makes topic-pattern practical for a "serving" function that must vary its behavior by the concrete topic a message arrived on, even though every matched topic shares one flow:

# in the first task of a topic-pattern flow, passed straight to the composable function
input:
  - 'input.metadata.topic -> topic'      # e.g. 'events.de' - the function decides per-topic behavior
  - 'input.metadata.partition -> partition'
  - 'input.body -> body'
process: 'topic.aware.dispatcher'

model.* is only needed when a later task (not the one receiving the message) needs the value — store it once ('input.metadata.topic -> model.source_topic') and reference model.source_topic from there on.

Second-level routing¶

Direct routing sends every record of a binding to one flow. When one topic carries mixed event types (a common Kafka pattern — e.g. a type header distinguishing orders from shipments), second-level routing picks the target per record instead: replace flow with a flows rule list (exactly one of the two, never both):

consumer:
  - topic: 'mixed-events'
    serializer: 'json'                 # optional; enables the input.body rule below
    ttl: '30s'                         # optional; deadline for task:// targets (default 30s)
    flows:
      - 'input.header.type(order) -> flow://order-flow'
      - 'input.header.type(order-*) -> flow://order-variant-flow'
      - 'input.header.type(regex: ^shipment-(eu|us)$) -> flow://shipment-flow'
      - 'input.body.event.kind(refund) -> task://v1.refund.processor'
      - 'default -> flow://catch-all-flow'

Each rule is <selector>(<matcher>) -> <target>, plus the mandatory default -> <target> fallback.

Selectors inspect one key-value of the inbound record:

input.header.<name> — a Kafka record header. The header name lookup is case-insensitive (Kafka preserves the producer's wire casing, so a rule must not depend on it); the value comparison stays case-sensitive.
input.body followed by a dot-bracket composite path — a Map body via input.body.order.type, a top-level List body via input.body[0].type, and any nesting of the two (input.body.items[1].kind). Body rules match only when the body is a Map or List — see payload prerequisites below.

Matchers — three modes, explicit over sniffing:

Form	Mode	Notes
`type(order)`	exact	case-sensitive value comparison
`type(order-*)`	wildcard	the presence of `` makes it one; each `` matches any run of characters
`type(regex: <expr>)`	regex	always explicit — the exception, not the norm

Wildcard and regex matchers use full-string matching (the topic-pattern precedent), so regex: shipment does not match my-shipment-1.

Evaluation. Order matters: the first matching rule wins, in declaration order — put the most specific rule first. A missing header/key, a non-Map body for an input.body rule, or a non-String value is a non-match, never an error; when no rule matches, default decides.

Targets:

flow://<flow-id> — dispatch to an Event Script flow exactly as direct routing does: same dataset, same model.cid seeding, same trace continuity, same flow ttl.
task://<route> — invoke a registered composable function directly, for processing simple enough that a flow is overweight. No input/output data mapping: all inbound record headers are copied to the function's input headers, the whole payload (byte[] or decoded Map) is the body, and trace context plus the business correlation-id propagate exactly as on the flow path (the function reads PostOffice.getMyCorrelationId() as usual). There is no metadata map on this path — a function that needs the record's envelope facts should be fronted by a flow instead. A bare function has no flow ttl, so the binding's optional ttl (duration syntax: 30s, 5m; default 30s) is the invocation deadline.

Both target kinds sit in the unchanged reliability envelope: unless auto-commit is on, the offset commits only after the selected flow or task finishes successfully, and a failure follows the same bounded-retry-then-dlq-topic path. A routing non-match is not a failure — it selects default.

All rules are validated at startup, fail-fast: every rule must parse (regexes compile; body keys use the dot-bracket composite-path convention), exactly one default is required, every flow:// target must be a compiled flow, and every task:// target must be a registered route (functions preload before the adapter starts) other than the flow engine itself — dispatch flows with flow://, never task://event.script.manager.

Payload prerequisites and `serializer: 'json'`¶

input.body.* rules need a Map body. On a schema.enabled binding the Confluent decode already yields one. For a registry-less topic (not every installation uses a schema registry), the optional per-binding serializer: 'json' tells the adapter to try deserializing each record value with the default SimpleMapper before routing:

a JSON object becomes a Map — input.body.<key> rules match, and the selected flow/task receives the decoded Map;
a JSON array becomes a List — addressable by bracket rules (input.body[0].type(order)) and delivered as decoded;
anything else — a scalar, or malformed text — keeps the raw byte[], which simply passes to the selected target. There is no special poison handling in the adapter: a target that cannot digest the bytes fails normally into the retry/DLQ path, while a default target designed for raw bytes handles them directly.

serializer is mutually exclusive with schema.enabled (the registry owns that decode) and is useful on a plain flow binding too — the flow receives a Map body without a schema registry. The parameter is open-ended for later extension; json is the only supported value today. Numeric values follow the customized-Gson semantics (integers arrive as Long — use util.str2int/util.str2long in a flow when a specific width matters).

Consumer group¶

group is the Kafka consumer group id, used exactly as given. Enterprise DevSecOps teams typically provision topics, ACLs, and consumer groups administratively, so the library never decorates the value. For a literal topic it defaults to kafka-flow-adapter.<topic> for convenience in dev/test; a topic-pattern binding has no sensible default (a regex string is not a group id) and must set group explicitly. All instances that share a group load-balance that binding's partitions; set it explicitly to your assigned group in production.

Partition pinning¶

When partition is present, the consumer manually assigns that single topic-partition instead of joining the consumer group for dynamic assignment. This bypasses group rebalancing — the pinned consumer reads exactly that partition — so you own the deployment model (one consumer per partition, or each pod pinning a distinct partition via partition: ${POD_PARTITION}). Offsets still commit under the configured group. Omit partition for normal group-managed consumption. Mutually exclusive with topic-pattern (below), since manual assignment needs concrete topic-partitions up front.

Pattern subscription¶

Set topic-pattern instead of topic to subscribe to every topic matching a regex, using Kafka's native subscribe(Pattern): the client tracks which topics currently match and adds/removes them from the subscription automatically as matching topics are created — no adapter-side polling of topic metadata, no restart needed when a new matching topic appears. All messages from every matched topic route into the same flow.

  - topic-pattern: 'events\.[a-z]{2}'   # matches events.de, events.fr, events.us, ...
    flow: 'process-region-event'
    group: 'region-events-group'        # required - no sensible default for a regex string

Two rules follow from this: topic-pattern cannot be combined with partition (manual assignment needs concrete topic-partitions up front, which a pattern doesn't provide), and group must be set explicitly. dlq-topic, if configured, must not itself match the pattern (see reliability).

Kafka client configuration¶

Connection and security settings live in template files, not code, because enterprise Kafka varies widely (on-prem, cloud, SaaS, Confluent; SASL/PLAIN, SASL/SCRAM, OAuth2, mTLS):

kafka-producer.properties — used by the publisher and the dead-letter writer.
kafka-consumer.properties — base config for every adapter consumer.
schema-registry.properties — the Confluent Schema Registry client (see registry authentication).

By default, each is loaded by ConfigReader from the bundled classpath template. Set kafka.producer.properties, kafka.consumer.properties, or schema.registry.properties only when you want a different location. A single location is normal; a comma-separated list is an optional fallback chain, useful when CI/CD renders an external file into a deployment volume and you still want to fall back to the bundled classpath template. All template values support ${ENV_VAR:default} substitution. The library pins only the parameters its contract depends on and lets the template own everything else:

Concern	Pinned by the library	From the template
Serialization	key=`String`, value=`byte[]` (de)serializers	—
Delivery semantics (consumer)	`enable.auto.commit` / `max.poll.records` — per-binding overlay (see delivery mode)	`auto.offset.reset`
Partitioning (producer)	`partitioner.class` defaulted (not pinned) to `SimpleRandomPartitioner`	any `partitioner.class` set here wins
Connection / security	—	`bootstrap.servers`, `security.protocol`, `sasl.`, `ssl.`, `acks`

bootstrap.servers is template-only via ${KAFKA_BOOTSTRAP_SERVERS:127.0.0.1:9092}, and the shipped consumer template sets auto.offset.reset=${KAFKA_AUTO_OFFSET_RESET:earliest} — a brand-new consumer group starts from the beginning of the topic; committed offsets govern thereafter. The byte[] wire contract keeps the building blocks serializer-free; richer encodings layer on top via the Schema Registry integration (JSON Schema / Avro), opt-in per binding.

Why a random partitioner? Kafka's default is a sticky partitioner — throughput-friendly, but at low volume it lands everything on one partition, leaving a multi-instance consumer group mostly idle. The library's SimpleRandomPartitioner is perfectly stateless and distributes keyless records uniformly. Records with an explicit partition header bypass it, and keyed records keep Kafka's murmur2 key-hash mapping. Set partitioner.class in kafka-producer.properties to override.

OAuth token URLs are allow-listed automatically. The Kafka client refuses to fetch an OAuth 2.0 token from a URL that is not on the JVM allow-list (the org.apache.kafka.sasl.oauthbearer.allowed.urls system property) — applications used to need a manual System.setProperty at startup. The library now registers every token endpoint it finds in the templates (sasl.oauthbearer.token.endpoint.url, bearer.auth.issuer.endpoint.url) on that allow-list before building a client, merging with — never clobbering — anything the operator set by hand.

Reliability: delivery mode, retry, and dead-letter¶

Delivery mode¶

By default (auto-commit: false, or omitted) the consumer commits offsets only after the flow finishes a message, one message at a time (max.poll.records defaults to 1). If the instance crashes before the commit, Kafka redelivers to a surviving instance in the group — the deliberate resilience-over-throughput trade-off.

Set auto-commit: true on a binding to trade that guarantee for throughput: Kafka commits offsets on its own periodic timer regardless of processing outcome, and max.poll.records defaults to 500 (still overridable via max-poll-records). A message being processed when a pod dies may already be considered committed and is not redelivered. Retry/dead-letter handling on flow failure is unaffected either way — auto-commit only changes when Kafka considers the offset committed, not whether a failure is retried or dead-lettered. Choose this per binding for high-volume topics (e.g. clickstream/telemetry) that can tolerate occasional loss on crash in exchange for throughput; leave strict topics on the default.

A flow succeeds when it replies with a status below 400 (any 2xx/3xx). A 4xx/5xx status — or a thrown exception, including a timeout when the flow does not reply within its own ttl — is a failure. (Kafka is asynchronous, so unlike an HTTP entry the adapter has no inherent request timeout: the flow's ttl is the processing deadline. There is no separate flow-timeout knob.)

Retry and dead-letter¶

On a failure, the message is retried up to kafka.flow.max.retries times (with kafka.flow.retry.backoff.ms between attempts), then written to the binding's configured dlq-topic:

One DLQ topic per binding, not per concrete topic — a topic-pattern binding that matches many topics still has a single dlq-topic (or none). The same flow that consumes a matched topic can reprocess a dead-lettered message later regardless of which concrete topic it originated from; that provenance is preserved via the dlq.origin.topic header, so a shared DLQ isn't the "mixing source schemas" anti-pattern it would be for unrelated topics. dlq-topic must not equal the source topic, nor match topic-pattern, or a dead-lettered message would be re-consumed by the same binding and fail forever — the adapter rejects that configuration at startup.
dlq-topic is optional. When omitted, a message that exhausts retries is dropped with a logged ERROR instead of being dead-lettered — the same fallback used when the DLQ write itself fails (below).
The DLQ write is confirmed (it blocks on broker acknowledgement, bounded by kafka.dlq.timeout.ms); on success the offset commits (or, in auto-commit mode, is left to Kafka's own timer as usual).
DLQ topics must be pre-provisioned (Kafka auto-creation is off in production). The original record's headers are preserved, plus dlq.origin.topic and dlq.error.

When there's no DLQ, or the DLQ write itself fails (data loss). A failed write to the DLQ is an exception of an exception with no further fallback. Blocking the partition to retry forever would re-run the failing flow and re-attempt the failing DLQ write indefinitely — a self-sustaining recovery storm (a known cause of prolonged outages). So the adapter instead logs a loud ERROR and commits (in manual-commit mode), deliberately dropping that one message to keep the partition live. This is a conscious data-loss trade-off; a planned improvement is a classic resilience alternative path — persisting the record to a durable store for later replay instead of dropping it.

Reprocessing (read the DLQ topic → fix → replay) is business-domain logic and is intentionally out of scope: the library guarantees durable capture (when a dlq-topic is configured and reachable), not replay.

Consumer liveness: rebalances and the processing deadline¶

Two robustness behaviors keep a binding alive through the realities of consumer-group life:

The poll loop survives transient consumer exceptions. A group rebalance (scale-out, pod churn) routinely makes an in-flight offset commit throw CommitFailedException or RebalanceInProgressException. The loop logs a WARN and continues — the uncommitted records simply redeliver to whichever consumer owns the partitions after the rejoin, preserving at-least-once delivery (flows must be idempotent, as always). Kafka's own retriable errors get the same treatment. Any other unexpected exception keeps the binding alive too, with an escalating pause (1s doubling to 30s) and an ERROR per occurrence — loud but alive, instead of a consumer thread that dies silently until the pod restarts.
max.poll.interval.ms is derived from the binding's worst-case processing time. Message processing happens on the poll thread (the flow's ttl is the deadline), so the worst case between two polls is the full retry envelope — (kafka.flow.max.retries + 1) × the slowest reachable flow/task ttl + retries × backoff — times max.poll.records, plus headroom. If that exceeds Kafka's max.poll.interval.ms (default 5 minutes), the group coordinator evicts the consumer mid-processing and the subsequent commit fails. The adapter therefore computes the envelope per binding at startup and raises max.poll.interval.ms to cover it (never lowering it below the Kafka default; the derivation is logged). An explicit max.poll.interval.ms in the consumer template is an operator decision and is respected as-is — with a WARN when the computed envelope exceeds it. Raising the interval is low-risk: a crashed pod is still detected by heartbeats (session.timeout.ms; broker-side group configuration under the KIP-848 consumer protocol); this setting only bounds time between polls.

Consumer rebalance protocol (KIP-848)¶

Kafka's classic rebalance protocol is client-driven with a group-wide synchronization barrier: when cloud infrastructure interrupts one pod, every member of the group stops, rejoins, and re-syncs — and a flapping member repeats that storm. The KIP-848 consumer rebalance protocol (GA since Apache Kafka 4.0) moves coordination to the broker's group coordinator and makes reassignment fully incremental: only the interrupted member's partitions move, survivors keep consuming. On clusters that support it, this materially reduces rebalance time and the CPU churn of unscheduled rebalances.

The protocol is selected per cluster in kafka-consumer.properties via group.protocol:

Value	Behavior
(unset) / `classic`	Kafka's classic protocol — works on every broker.
`consumer`	The KIP-848 protocol, unconditionally. Fails at runtime if the cluster does not support it.
`auto`	The adapter probes the cluster once at startup and picks `consumer` when available, `classic` otherwise.

The bundled kafka-consumer.properties sets group.protocol=${KAFKA_GROUP_PROTOCOL:auto}, so auto is the default: a KIP-848 cluster gets the incremental protocol without configuration, an older cluster keeps classic. Override it with the KAFKA_GROUP_PROTOCOL environment variable or in your own template. (The Rust engine ships the same default; its auto starts with consumer and falls back to classic at the first join when the broker refuses the protocol, since its client has no feature probe.)

How auto decides. KIP-848 enablement is a finalized feature flag (group.version >= 1) — controller-managed and cluster-wide, so it is authoritative even during a rolling broker upgrade. The probe reads it via the ApiVersions handshake that every Kafka client performs on connect: the broker answers it before authentication completes and never applies an ACL to it, so the probe needs no grant beyond the connection credentials already in the template. One probe per cluster per application instance; the decision is stated in the startup log. Any probe failure — an older broker, an unreachable cluster, a Kafka-compatible endpoint that does not report features — resolves to classic, the safe default.

Client tuning that conflicts. Under the consumer protocol, session.timeout.ms, heartbeat.interval.ms and partition.assignment.strategy move to broker-side group configuration — a client that sets them together with group.protocol=consumer fails fast with a ConfigException. When the template sets any of them, auto therefore resolves to classic with a WARN naming the conflicting keys: remove them to let auto upgrade. Everything the adapter itself manages — group.id, the delivery-mode overlay, the derived max.poll.interval.ms — is valid under both protocols.

Prerequisites and managed services. The cluster must run Apache Kafka 4.0+ with the group.version feature enabled (new 4.0+ clusters enable it at format time; upgraded clusters enable it explicitly — check with kafka-features.sh describe). Confluent Platform 8.x carries the Apache 4.x core and reports the flag; for other managed or Kafka-compatible services (Confluent Cloud, AWS MSK, Azure Event Hubs), verify against your actual cluster — wherever the flag is not reported, auto simply keeps classic. Migration is online: a group converts when members join with the consumer protocol (mixed members interoperate during a rolling deploy) and reverts if all new-protocol members leave. To force the classic protocol regardless of cluster support, set group.protocol=classic explicitly.

Shutdown: leaving the group¶

On JVM shutdown — SIGTERM from an orchestrator's rolling restart, or Ctrl-C — the flow adapter closes every binding's consumer before the process exits, and the consumer's close sends the group coordinator a LeaveGroup: the member's partitions are reassigned to the surviving members at once. Without that, the broker only notices the dead member when its session expires — 45 seconds by default under the KIP-848 consumer protocol — and every partition it held sits unread for that long, which on a rolling deploy is a pause of the same length for the pod's share of the traffic. The shared producer is closed after the consumers, waiting up to the same ten seconds for its buffered records to be acknowledged; when the grace ends first, the records still unacknowledged are failed and the log names how many (Kafka producer closed after 10 s grace - N message(s) undelivered) — a stopping pod must not wait on a dead broker past its termination grace. The whole sequence is KafkaRuntime.shutdown(), registered on the platform's shutdown lifecycle (Platform.onShutdown) when the clients open; it is idempotent and safe when nothing was started. The log confirms each step — Kafka flow consumer for <topic> closed - left group <group>, then Kafka producer closed - buffered records delivered — and the broker's own log shows the member leaving instead of being fenced. A consumer mid-flow is given ten seconds to finish its current record before the close is forced, so a stuck flow cannot hold the shutdown hostage. The Rust port shuts down the same way, its producer flushed within the same grace.

Outbound: publishing to Kafka¶

simple.kafka.notification is a composable function that publishes an event to a topic. Send it an EventEnvelope with a topic header (required), an optional partition header (see partitioning strategies), a body, and any other headers (forwarded as Kafka headers):

po.send(new EventEnvelope().setTo("simple.kafka.notification")
        .setHeader("topic", "outgoing-events")
        .setHeader("cid", businessCorrelationId)
        .setBody(payloadBytes));

The body is byte[] (published verbatim — the minimalist default), or a Map/List, automatically serialized to JSON bytes — the outbound symmetry of the inbound serializer: 'json': the producing application writes a Map, the wire carries JSON bytes, and a consuming binding with serializer: 'json' hands its flow a Map again. This JSON convenience applies to non-schema-registry topics only; null stays null (a Kafka tombstone).

Publishing is drop-n-forget (Kafka's commit log is the durable buffer), but async delivery failures are logged rather than silently masked.

Two contract details worth knowing: any other body type (a String, a PoJo) is rejected loudly with an IllegalArgumentException — convert to byte[] or a Map/List first. And the correlation-id header is auto-stamped as a fallback: when the flow maps no value under the configured header (default cid), the publisher stamps the flow's own business correlation id (model.cid); an explicitly mapped value always wins. With a customized kafka.correlation.id.header, map to the configured name — a header.cid mapping under a custom name is forwarded as a literal cid record header, never renamed.

One header opts a publish into the Confluent wire format instead of raw byte[]: subject (with an optional version; see Schema Registry). It is an encoding directive — consumed by the function, not forwarded as a Kafka header. On this schema path the body must be byte[] (a pre-serialized JSON document) — passing a Map or List is rejected with IllegalArgumentException; the Map/List JSON convenience applies to non-schema topics only.

Keep the worker pool small. When the Schema Registry is in use, simple.kafka.notification runs on kernel threads (@KernelThreadRunner) because Confluent's serializers are not thread-safe. Each worker instance owns its own encoder and is single-flight, so a small pool (the default is 5) sustains high throughput — Kafka publishing is fast and mostly waits on the broker acknowledgement. Raise instances only if profiling shows the publishing path is the genuine bottleneck. This constraint does not apply to raw byte[] publishing (no Schema Registry).

Partitioning strategies¶

Three mechanisms decide which partition an outbound message lands on, in precedence order:

Explicit partition header — the caller (usually a flow's data mapping) names the target partition and every partitioner is bypassed. This is the building block for content-based partitioning (below).
A custom partitioner in the producer template — the externalized kafka-producer.properties may set Kafka's standard partitioner.class; the library registers its default with putIfAbsent, so a template's own value always wins. The Kafka Partitioner API receives the record's key and value (payload inspection works — plain JSON bytes on non-schema topics, Confluent-framed bytes on schema topics), but not the record headers — header-based partitioning is impossible at this layer, a Kafka API limitation.
SimpleRandomPartitioner — the library default for unkeyed records: simple random sampling spreads low-volume traffic evenly across partitions (Kafka's own sticky default batches onto one partition, which starves multi-instance consumer groups at low volume). Keyed records keep Kafka's murmur2 hashing.

Content-based partitioning — the composable pattern. When the partition must be derived from a record header or a payload key-value (a tenant, an entity id), compute it in the flow — where the whole record is visible — and pass the explicit partition header. A tiny selector function plus data mapping, no client plumbing:

tasks:
  - input:
      # any header can drive the decision - including ones a Kafka Partitioner could never see
      - 'input.header.x-routing-value -> header.routing-value'
      - 'input.body -> *'
    process: 'partition.selector'          # e.g. hash(routing-value) % partition count
    output:
      - 'result.partition -> model.partition'
      - 'result.payload -> model.payload'
    description: 'Derive the target partition from the record content'
    execution: sequential
    next:
      - 'simple.kafka.notification'

  - input:
      - 'text(outgoing-events) -> header.topic'
      - 'model.partition -> header.partition'   # explicit partition bypasses all partitioners
      - 'model.payload -> *'
    process: 'simple.kafka.notification'
    output: []
    description: 'Publish to the selected partition'
    execution: end

A deterministic selector (same value → same partition) gives per-entity ordering — the classic reason for content-based placement. This is the outbound mirror of second-level routing: content inspection expressed in the application layer, where it is legible and governable and can see the whole record, rather than buried in client configuration.

Trace continuity across Kafka¶

Rather than forwarding the caller's stale traceparent, the notification function stamps a fresh W3C traceparent from its own current span; the adapter parses it on the way in and chains the flow onto that span. The result is one continuous distributed trace across the asynchronous Kafka boundary — the two notification hops are the bridge spans. See Observability.

Health check¶

The library ships a ready-made health-check function at route kafka.health (auto-registered when the jar is on the classpath). Opt in by listing it as a health dependency in application.properties:

mandatory.health.dependencies=kafka.health
# or, when Kafka should be reported but not fail /health:
# optional.health.dependencies=kafka.health

The probe is deliberately minimal: one Kafka Metadata request (KafkaConsumer.listTopics) using the module's consumer template - it joins no consumer group, commits no offsets, and needs no admin privileges. The Metadata request itself requires no ACL: brokers filter the response to the topics the principal may Describe rather than rejecting the request, so under a fully locked-down principal the probe still succeeds (with a visible topic count of 0) - the successful round trip proves connectivity, TLS/SASL authentication, and a served API request. A reachable cluster reports a status map (including the visible topic count, which may be 0 under restrictive ACLs); an unreachable one fails the check with a 503 status and a key-value message (text for the DevOps reader, code for the status code), so /health marks the dependency down and the endpoint answers non-2xx while the application is DOWN.

During application start-up the check returns a placeholder healthy status while the Kafka client warms up in the background - /health neither fails nor blocks before the client and the rest of the start-up sequence complete. After the first successful probe, or once the grace period expires, every check is live. Two keys tune the behavior: kafka.health.timeout (default 5s) and kafka.health.startup.grace (default 30s) - see the Configuration Reference.

The probe's client configuration is resolved lazily - when the probe client is built, and again whenever a failed probe forces a rebuild - never at construction time. kafka.health is registered before your @MainApplication runs, so a bootstrap that fetches secrets and publishes them as system properties (the vault pattern) has not executed yet - a config template frozen at construction would interpolate such a credential as missing and fail every probe from then on. With lazy resolution the first probe after the credential lands simply succeeds; nothing needs a restart. While the template is still incomplete - the client cannot even be built from it - type=health reports a passing Waiting for Kafka connection status rather than a failure: failing /health would invite the container orchestrator to restart the pod, and a restart cannot produce the credential. A real connectivity failure (client built, cluster unreachable) fails the check with status 503. The same applies to secondary.kafka.health.

Waiting is only for a value that has not landed yet — a broken template fails. The leniency above exists for exactly one situation: a credential a later bootstrap will publish. A template referring to a class that is not on the classpath will never become usable by waiting, so type=health fails with 503 and says Kafka client configuration is unusable - <reason> rather than reporting healthy. The distinction matters because the passing-while-broken case is genuinely hard to spot: a field deployment logged an unresolvable deserializer class every five seconds for hours while /health reported healthy throughout. The message deliberately names the configuration rather than the network, so the reader looks at the classpath instead of the cluster.

Class-valued settings are supplied as class objects, not names. Kafka resolves a class name through the thread context classloader, falling back to its own loader only when that is null — so a non-null but wrong context loader turns a present class into Class ... could not be found. That is what bit the field: kafka.health is a @KernelThreadRunner and builds its client on a pooled kernel thread, while the flow adapter's consumers — same config, same jar, ordinary threads — were fine. The module now puts Class objects into the client properties and the probe pins the thread context loader for the duration of the whole probe, so nothing it does depends on which thread happens to run it. Nothing to configure; a template that names its own partitioner, assignor or interceptors still wins as before.

Why the whole probe, and not just the client construction. Kafka consults the context loader before any client exists. A Type.CLASS setting's default is resolved the moment the key is defined, inside the static initializer of Kafka's own config classes — and sasl.oauthbearer.jwt.retriever.class defaults to a class name. So merely initializing ConsumerConfig is a classloading event, with no broker, no SASL and no credentials involved. Two things make that worse than an ordinary lookup failure: it surfaces as an ExceptionInInitializerError (an Error, which a catch (Exception) does not hold, so /health answered a raw 500 instead of a 503), and a class whose initializer threw stays erroneous for the life of the JVM — every later touch fails on every thread, however correct its loader. The probe therefore has to get the loader right the first time; waiting and retrying cannot help. Fixed in 4.12.12, which is why 4.12.11 still failed on a produce-only leg: that path resolves its template through ConsumerConfig.configNames() before building anything, whereas the consumer path's references to the same class are compile-time constants the compiler inlines.

On a produce-only leg the probe uses the producer template. With kafka.consumer.enabled=false there are no consumer credentials to build a probe from, yet a bridge is healthy only when both clusters are reachable. So the probe follows whichever client the deployment configured, reading connection and security settings (bootstrap.servers, security.protocol, sasl.*, ssl.* - named identically in both client surfaces) from kafka-producer.properties. Producer-only settings such as acks are filtered out rather than logged as unknown config. Nothing else about the probe changes; it still joins no group and needs no ACL.

Dual-cluster applications get a twin for the second cluster: twin-kafka ships secondary.kafka.health, so a bridge lists both dependencies.

Schema Registry: typed payloads (opt-in)¶

The default wire contract is raw byte[], which keeps the building blocks serializer-free. To interoperate with existing Confluent client projects, the library can also speak the Confluent Schema Registry wire format — [magic 0x00][4-byte global schema id][payload] — using Confluent's own serializers as a library (not a reinvented codec). JSON Schema and Avro are supported.

Protobuf is not currently supported. It was implemented and demoed in an earlier development phase but removed before its first release: Confluent's kafka-protobuf-provider depends on com.squareup.wire:wire-runtime-jvm, a discontinued artifact carrying an unpatched denial-of-service CVE (CVE-2026-45799 / GHSA-7xpr-hc2w-34m9) with no fix available anywhere in that coordinate — Wire's maintainers will not patch it, and Confluent has not adopted the renamed wire-runtime replacement as of kafka-protobuf-provider:8.3.0. This is a tracked backlog item, not an abandoned one: it gets re-wired once Confluent moves, or sooner for a specific field installation that explicitly needs Protobuf and accepts the residual risk. SchemaType.PROTOBUF is still recognized internally so a misconfigured attempt fails clearly (UnsupportedOperationException), not silently.

Set schema.registry.url to turn the feature on (point it at a real Confluent registry or the local schema-registry-standalone mock). When it is unset, schema features stay off and the library keeps its raw byte[] behavior.

schema.registry.url=${SCHEMA_REGISTRY_URL:http://127.0.0.1:8081}
schema.registry.cache.ttl=30m                          # TTL for the in-memory schema cache (by id)

Registry authentication (OAuth 2.0 / basic)¶

The registry client's connection and security parameters live in the schema-registry.properties template (same mechanics as the producer/consumer templates: classpath by default, optionally overridable with schema.registry.properties). Everything in it is passed verbatim to the Confluent Schema Registry client, so any client parameter — including optional, installation-specific ones such as bearer.auth.logical.cluster and bearer.auth.identity.pool.id — works without a library change.

OAuth 2.0 client-credentials (e.g. Azure AD / Entra ID):

bearer.auth.credentials.source=OAUTHBEARER
bearer.auth.issuer.endpoint.url=${SCHEMA_REGISTRY_OAUTH_TOKEN_URL:}
bearer.auth.client.id=${SCHEMA_REGISTRY_CLIENT_ID:}
bearer.auth.client.secret=${SCHEMA_REGISTRY_CLIENT_SECRET:}
bearer.auth.scope=${SCHEMA_REGISTRY_OAUTH_SCOPE:}

The Confluent client fetches the bearer token from the issuer endpoint with the client id/secret, sends it as Authorization: Bearer on every registry request, and caches it, refreshing shortly before expiry (bearer.auth.cache.expiry.buffer.seconds, default 300). The issuer URL is auto-registered on the JVM allow-list, so no manual System.setProperty is needed. Other credential sources come free with the pass-through: SASL_OAUTHBEARER_INHERIT (reuse the Kafka transport's SASL OAuth settings — one credential for broker and registry), STATIC_TOKEN (dev/test), and basic.auth.credentials.source=USER_INFO for basic authentication. Keep secrets in environment variables via the ${ENV_VAR} substitution; the shipped template is fully commented out, so an unauthenticated registry (like the local mock) keeps working with zero configuration.

Produce: subject-driven¶

simple.kafka.notification serializes the body into the wire format when you supply a subject header:

Header	Description
`subject`	The registry subject to serialize against. The schema must be pre-registered; the producer resolves the subject to a global schema id (and its type — `JSON` or `AVRO`) from the registry and never registers.
`version`	Optional. The subject version to resolve: a positive integer to pin a specific version, or `latest` to track the current version. Defaults to `latest`.

# in a flow task that publishes via simple.kafka.notification
input:
  - 'text(orders) -> header.topic'
  - 'text(orders-value) -> header.subject'    # version omitted → latest
  - 'model.payload -> *'        # the body: must be byte[] (a JSON document) on the schema path
process: 'simple.kafka.notification'

The producer resolves the subject (+ version) to a global schema id and its schema type from the registry, then serializes with Confluent's own serializer. The wire format itself is unchanged — it still carries only the global id ([magic 0x00][4-byte global schema id][payload]) — and the consumer (id-from-wire) is unchanged; only the producer's input moved from an explicit id+type to a subject. Whoever registers the schema — CI, a client project, an admin tool — owns the subject naming strategy (TopicName / RecordName / TopicRecordName are all supported, and a topic can carry many record types). This assumes schemas are governed artifacts registered out-of-band, as they are in practice; the producer never auto-registers.

CSFLE (Client-Side Field Level Encryption)¶

minimalist-kafka supports Confluent CSFLE by delegation — the framework adds no encryption/rule logic of its own. A schema's ruleSet (its ENCRYPT rules, tagging fields for encryption) travels with the schema itself through the same getSchemaById lookup the producer/consumer already use, and Confluent's own serializer/deserializer run those rules during the same use.schema.id + envelope serialize/deserialize call this library already makes. So there is no separate "encrypted" code path to opt into — CSFLE activates the moment (a) the encryption executor + a KMS driver are on the classpath and (b) their configuration is present; everything else — resolving keys, encrypting tagged fields on write, decrypting on read — is Confluent's serializer/ deserializer doing exactly what it would do in any Confluent-based application.

1. Dependencies. system/minimalist-kafka/pom.xml declares:

io.confluent:kafka-schema-registry-client-encryption — the field-encryption rule executor (auto-discovered via ServiceLoader; no explicit rule.executors config needed). This is a normal compile dependency, so an app that depends on minimalist-kafka inherits the executor transitively — nothing to add.
Exactly one cloud KMS driver — which your application must supply itself. In this module's own POM the AWS driver (io.confluent:kafka-schema-registry-client-encryption-aws) is uncommented as the default/template (Azure and GCP are present alongside it, commented out). But that dependency is marked Maven <optional>true</optional>, so it is not inherited transitively by a downstream consumer of the published artifact. Your app must therefore declare exactly one KMS driver appropriate to its environment — AWS, Azure, GCP, or (for tests) the local Tink driver — on its own classpath. A field installation configures one KMS vendor, not several.

Why optional rather than a default AWS dependency for everyone? Forcing the AWS SDK onto every consumer would be wrong for Azure/GCP and non-CSFLE users. So minimalist-kafka ships the vendor-neutral executor and lets the application pick its one KMS driver. A common symptom of skipping this: a subject configured with a CSFLE rule fails at runtime because no KMS driver is on the app's classpath.

2. The ENCRYPT rule — and its KEK/KMS identity — is per-subject, set on the schema, not in this app's config. When a subject is registered with an ENCRYPT rule tagging a field (e.g. confluent:tags: ["PII"] inline in the schema, or via a Metadata tags map), that rule's own parameters — encrypt.kek.name/encrypt.kms.type/encrypt.kms.key.id — say which key encrypts that subject's tagged fields. This is deliberate: Confluent's rule executor resolves these from the registered rule (or the schema's Metadata) and never from this library's serde config, so different subjects can use different KEKs/vendors without any code or config change here. Whoever registers/governs the schema — CI, an admin tool — owns this binding. A schema with no ENCRYPT rule serializes exactly as before (plaintext); CSFLE is per-subject, never a single global on/off switch.

Registry authentication is inherited by the serdes. The serializer/deserializer's own registry and DEK-registry clients receive the same schema-registry.properties authentication this library uses for its schema lookups (with any schema.registry.serde.* overrides applied on top) — a governed registry that returns 401 to anonymous clients works without duplicating credentials.

3. What does go in application.properties — the schema.registry.serde.* pass-through. This is reserved for genuinely global, app-level settings the KMS driver itself needs (not per-subject key identity): typically nothing at all if you rely on your cloud's default credential chain (e.g. an IAM role for AWS), or explicit driver credentials if you don't:

# Only needed if you are not relying on the AWS SDK default credential chain (an IAM role, etc.):
schema.registry.serde.access.key.id=${AWS_ACCESS_KEY_ID}
schema.registry.serde.secret.access.key=${AWS_SECRET_ACCESS_KEY}

Any property under this prefix is merged, prefix stripped, into both the serializer's and the deserializer's Confluent config map (decrypt is symmetric, so both directions need the same driver credentials). It is a generic pass-through — a KMS driver's own config keys (AWS's access.key.id/ secret.access.key/profile/role.arn, or Azure's/GCP's equivalents) flow through with no code change here.

4. Wire format and consumer are unchanged. The frame is still [magic 0x00][4-byte global schema id] [payload] — CSFLE only changes the value of the tagged fields inside that payload to ciphertext (with embedded DEK metadata Confluent's deserializer reads to decrypt). DLQ handling, tracing, and the schema.enabled consumer binding are unaffected.

5. Not covered by the standalone mock. schema-registry-standalone is a plaintext dev tool with no ruleSet/KMS support (deliberately — see its own docs). Test/demo CSFLE against a real Confluent Schema Registry and a real (or local) KMS.

A separate registry identity for the consumer side¶

By default one codec — one Schema Registry identity, taken from schema-registry.properties — serves both directions: simple.kafka.notification encodes and the flow adapter decodes with it. Some Confluent installations grant a service's registry access per direction — most visibly for CSFLE, where key (KEK) access comes through separate produce and consume identity pools — so no single identity can decrypt everything the service consumes, and a message decoded under the wrong identity fails as a poison message (dead-lettered) with no configuration remedy. Set schema.registry.consumer.properties to give the flow adapter its own codec, built under the schema.registry.consumer key prefix against the same schema.registry.url:

# Producer identity (unchanged)
schema.registry.url=${SCHEMA_REGISTRY_URL}
schema.registry.properties=classpath:/schema-registry.properties

# Consumer identity: the SAME template, plus the one override that differs
schema.registry.consumer.properties=classpath:/schema-registry.properties
schema.registry.consumer.serde.bearer.auth.identity.pool.id=${SCHEMA_REGISTRY_CONSUME_POOL_ID}

It is the prefix seam twin-kafka uses for a second cluster's registry, applied here to one registry with two identities (twin-kafka's secondary cluster has the same opt-in under secondary.schema.registry.consumer.properties):

Unset or blank = unchanged. The adapter shares the producer's codec exactly as before, and the ${ENV_VAR:} idiom (blank when the variable is unset) keeps it that way per environment; there is no separate switch to remember. The opt-in does not turn schema features on — schema.registry.url stays the switch.
The consumer keys derive from the prefix: schema.registry.consumer.properties (the template — reuse the producer's file or point at a second one), schema.registry.consumer.serde.* (pass-through overrides on top of that template) and schema.registry.consumer.cache.ttl (its own caches). The registry URL is shared: a consumer decodes messages whose ids were minted by the registry its producers use.
Where an override lands. A schema.registry.consumer.serde.* entry reaches the Confluent deserializer's configuration — and through it the DEK-registry client CSFLE builds from that configuration, where key access is decided. The codec's own schema-by-id lookups still authenticate with the template's identity; when the consume identity must cover those too, point schema.registry.consumer.properties at a second template that carries it instead of layering an override.
The consumer codec reads only its own prefix. A schema.registry.serde.* KMS driver credential (item 3 above) that the producer needs must be repeated under schema.registry.consumer.serde.* — the two codecs are configured independently.

A commented sample of both variants ships with the sync-over-async demo: its application.properties carries the opt-in block, and schema-registry-consumer.properties next to it is the second-template form. The local registry helper enforces no identity pools, so the sample shows the wiring; the registry's own access grants decide the boundary.

Consume: decode by embedded id¶

Set schema.enabled: true on a consumer binding. The adapter reads the magic byte + embedded id, looks up the registered schema's type, dispatches to the matching deserializer, and hands the flow a Map as input.body (instead of byte[]). No flow-YAML change is needed (input.body -> * is type-neutral); a schema-fed flow task simply takes Map<String,Object> instead of byte[].

consumer:
  - topic: 'orders'
    flow: 'process-order'
    group: 'order-group'
    schema.enabled: true

A decode failure is a poison message (retrying won't help), so the raw record is dead-lettered immediately via the DLQ path rather than retried.

Notes¶

One subject-driven path, two formats. The producer and consumer are type-generic; only the subject (and the registered schema behind it) differ — the producer reads the schema type from the registry, so the flow never names it. JSON Schema is open (additionalProperties), while Avro records are closed-shape — a message must match the declared fields, and a non-schema field is dropped on the wire. Avro applies declared field defaults for absent fields, and decodes to a generic record (no generated classes), rendered to a Map.
Schema cache. Lookups by id are cached in memory (platform ManagedCache, TTL schema.registry.cache.ttl, default 30m) to cut registry round-trips. A global schema id is immutable, so a cache hit is always the right schema. Positive results only — a not-found id is never cached, so a schema registered while the app is running becomes visible on the next lookup. The TTL lets schema changes be picked up without restarting pods (handy in dev / lower environments); lengthen it in production where schemas change rarely. The cache is rebuildable and cleared at startup.
Subject→id resolution cache. The producer also caches the subject (+ version) → schema id resolution, and how long depends on the version. A pinned numeric version (subject + version: N) maps to one immutable schema id, so it is cached long (effectively forever). latest (the default) can change when a new version is registered, so it is cached on a short TTL and re-resolved frequently, picking up a new current version without a restart. Pin a version in production paths where the schema must not shift underneath you; use latest in dev / lower environments where tracking the newest schema is convenient.
Worked example. The sync-over-async demo runs the same end-to-end flow over both formats (json-topic-1/2, avro-topic-1/2) alongside the raw byte[] path.

Configuration keys¶

All keys are documented in the Configuration Reference. The essentials:

Key	Default	Description
`yaml.kafka.flow.adapter`	—	Adapter config location; unset = inbound adapter off.
`kafka.producer.enabled`	`true`	Set `false` on a consume-only leg to build no producer — see switching off a client. A binding with `dlq-topic` then fails startup.
`kafka.consumer.enabled`	`true`	Set `false` on a produce-only leg to start no adapter consumer; `kafka.health` then probes through the producer template.
`kafka.producer.properties`	`classpath:/kafka-producer.properties`	Producer template location. Set to an external file path (or explicit fallback list) to externalize.
`kafka.consumer.properties`	`classpath:/kafka-consumer.properties`	Consumer template location. Set to an external file path (or explicit fallback list) to externalize.
`kafka.dlq.timeout.ms`	`10000`	Confirm-write timeout for the dead-letter publish. (Flow processing has no timeout knob — the flow's own `ttl` is the deadline.)
`kafka.flow.max.retries`	`3`	Retry attempts before dead-lettering.
`kafka.flow.retry.backoff.ms`	`500`	Pause between retry attempts.
`schema.registry.url`	—	Confluent Schema Registry URL; unset = schema features off (raw `byte[]`).
`schema.registry.properties`	`classpath:/schema-registry.properties`	Registry client template location — auth/SSL parameters passed verbatim to the Confluent client (see registry authentication). Set to an external file path (or explicit fallback list) to externalize.
`schema.registry.cache.ttl`	`30m`	TTL for the in-memory (`ManagedCache`) schema cache (by id); positive results only; cleared at startup.
`schema.registry.consumer.properties`	—	Opt in to a separate registry identity for the consumer side: the flow adapter's own registry client template (the producer's file or a second one), with `schema.registry.consumer.serde.*` overrides on top. Unset or blank = the adapter shares the producer's codec.