Configuration

Turn your Kafka records into the events your workflows expect: the two config blocks and why there are two, subscription, envelope, decoding, and mapping.

The Kafka connector reads one YAML file. This page is what is in it and what a record turns into on the way through. For the full key lists see the duraton: and kafka: references.

The two config blocks

The file is YAML and has exactly two blocks, split by who owns the vocabulary:

duraton:                          # Duraton's surface: we own, version and document these names
  app: billing-ingress
  topics: [payments.events]
  decoder: json
  envelope: wrapped
  mapping:
    eventName: value.type

kafka:                            # forwarded: Apache Kafka's own property names
  bootstrap.servers: kafka-1:9092,kafka-2:9092
  group.id: duraton-billing
  auto.offset.reset: latest
  isolation.level: read_committed
  security.protocol: SASL_SSL
  sasl.mechanism: SCRAM-SHA-512
  sasl.username: duraton

Two blocks, because there are two vocabularies and pretending otherwise costs you something either way. Anything about what a record becomes is ours: topics, decoding, the envelope, the event mapping, the filter. Anything about how a Kafka client behaves keeps Kafka's own property names, so your existing knowledge and your existing client.properties transfer unchanged, and nothing in your config is tied to which Go client sits underneath.

Subscription is the one thing that looks like it belongs in kafka: and does not. Neither Kafka vocabulary has a property for what a consumer subscribes to - in the Java client it is a method call - so topics, topicPattern, excludeTopics and assign live in duraton:.

Subscription

At least one of three, and topics cannot be combined with either of the others:

topics - a list of exact names.
topicPattern - one regular expression, plus optional excludeTopics.
assign - direct partition assignment, topic: {partition: offset}, for a consumer that joins no group. Mutually exclusive with kafka.group.id.

A pattern is matched against the whole topic name, which surprises people; see Behaviour worth knowing.

The envelope

A Kafka record carries as much outside its value as inside it. The default wrapped envelope lifts all of it into the event payload, so a path expression can reach any of it:

{
  "key": "A1",
  "headers": { "content-type": "application/json" },
  "topic": "payments.events",
  "partition": 3,
  "offset": 91422,
  "timestamp": "2026-07-29T10:00:00Z",
  "value": { "type": "payment.captured", "amount": 4200 }
}

envelope: value-only sends the decoded value alone, for payloads that already carry everything.

A mapping path always resolves against the wrapped structure, in both modes. value.orderId means the same thing whichever envelope is set; value-only changes what is sent, not what a path sees. So a value-only event can still be named from the topic or tagged from the record key even though neither reaches the payload.

Decoding

`decoder`	Behaviour
`json` (default)	The value is parsed as JSON.
`raw`	The value is passed through as a base64 string.
`text`	The value is passed through as a UTF-8 string; invalid UTF-8 is poison.
`avro-confluent`	Confluent framing is stripped and the writer schema is resolved from the Schema Registry, which the value cannot be read without.
`json-schema-confluent`	Confluent framing is stripped and the JSON beneath it is parsed. No Schema Registry is contacted.

A decode failure is always a poison outcome, never a silent drop. A decoder that could not reach what it needed - a Schema Registry that blinked - is not: that is a stall, because poisoning there would discard records over a dependency being briefly unavailable. A registry that refuses the credential stalls for the same reason: the records may be perfectly good, and only the configuration is wrong.

json-schema-confluent needs no registry because a JSON Schema value is already JSON once the framing is off. The framing is a short header naming the schema the value was written against - by schema id or by schema GUID, whichever the producer's serializer wrote. The message-index bytes that sit between the identifier and the data are a Protobuf feature only, which is why the three Confluent formats are not interchangeable and only this one decodes offline. A schema validates a value; it is not needed to read one. Confluent's own deserializer does not validate by default either, so neither does this.

A value that is not framed is poison, named as what it is rather than as a parse failure. Plain JSON sent to this decoder starts with {, which is byte 0x7b:

the json-schema-confluent decoder rejected the record value: its first byte is 0x7b, which is not a framing version (0x00 carries a schema id, 0x01 a schema GUID)

The contract also names protobuf-confluent. It is not built in this release, and it is refused by name rather than as an unknown value, so "not available yet" never reads as "not a decoder":

  - duraton.decoder "protobuf-confluent" is not available in this release: accepted values are json, raw, text, avro-confluent, json-schema-confluent
  - duraton.schemaRegistry.url is required when duraton.decoder is "protobuf-confluent"

The second bullet appears alongside the first whenever no registry address is set: a refused value is still held to the rest of its own rules, so one mistake is not allowed to hide another.

Avro and the Schema Registry

avro-confluent is the decoder the registry is load-bearing for. An Avro value carries no field names or types of its own, so the writer schema is not a validation of the bytes but the only thing that says what they mean. The identifier in the framing header is resolved against schemaRegistry.url, with schemaRegistry.auth sent as HTTP basic auth, and the answer is held for schemaRegistry.ttl so a partition costs one round trip per schema rather than one per record.

GUID framing needs a registry that answers GUID lookups. The two ways a producer can name the schema reach two different endpoints: a schema id through GET /schemas/ids/{id}, a schema GUID through GET /schemas/guids/{guid}. Confluent Schema Registry serves both. Redpanda's built-in registry serves only the first, so against it a producer must write schema-id framing - which is what Confluent's serializers do by default.

Avro's own JSON encoding wraps a union value in an object keyed by the branch's type name, so {"string": "acme"} rather than "acme". Duraton unwraps the ["null", T] case, because that is how Avro spells an optional field and the wrapper would otherwise put a type name in the middle of every path a mapping, a flow key or a filter reads:

# schema: {"name": "customer", "type": ["null", "string"]}
mapping:
  tags:
    customer: value.customer     # "acme", not value.customer.string

A union with genuine alternatives keeps its wrapper, because there the branch is part of what the value means. The workflow receives it in full:

{"dest": {"shop.Address": {"city": "Berlin"}}}

The rule is exactly the one Kafka Connect applies: a union of null and one other type is flattened to that type, and anything else keeps its branch.

A path cannot reach through a branch whose type is namespaced. Paths split on . and have no escaping, so the segment shop.Address reads as two segments and selects nothing - the same cost a header whose name contains a dot pays. The value is intact in the event data and a workflow reads it normally; it is only mapping paths, flow keys and dedupe-id templates that cannot address it. Where you need one as a tag, give the union's branch a schema with no namespace.

Mapping

One key per settable event field, so the mapping cannot drift from the event contract. Each value is a literal or a dotted path into the envelope.

Paths split on . only: no array indices, no escaping, and a missing path resolves to the empty string.

Every key is listed in the duraton: reference.

Filtering - drop records before they cost anything.
Delivery - when the offset moves.
duraton: reference - every key, type and default.
Diagnostics - see what your own records become.