[FEATURE] Kafka Operation Binding: `error-topics` for Retry and DLQ Topic Provisioning

[ivangsa]

Why do we need this improvement?

Motivation

Enable provisioning complete Kafka infrastructure from AsyncAPI definitions alone, making AsyncAPI the single source of truth, not just for the API contract but for the full cluster topology.

With channel bindings you can already provision main topics. But retry and DLQ topics have to be defined somewhere else. Today there are two options:

• Define them outside AsyncAPI. This breaks the single source of truth and causes API drift.
• Define them as channels and operations. A single receive operation with 5 retry topics and a DLQ would require 6 additional channels and 12 additional operations. And they should not be there: retry and DLQ topics are not part of a service's public API. They are internal infrastructure owned by one consumer.

This proposal closes that gap without touching the contract.

How will this change help?

This will allow teams define and provision error topics using AsyncAPI as the single source or truth in a non-verbose non-invasive manner.

ZenWave SDK supports generating Terraform from AsyncAPI files for provisioning Kafka OSS, Schema Registry or Confluent Platform and uses this as an extension to provision kafka error topics.

https://www.zenwave360.io/zenwave-sdk/plugins/asyncapi-ops/

Screenshots

No response

How could it be implemented/designed?

Proposed Design

Placement

At the operation binding level, under bindings.kafka. Retry and DLQ topics are a consumer concern, the producer has no involvement.

operations:
  onSomethingHappened:
    action: receive
    channel:
      $ref: '#/channels/some-channel'
    bindings:
      kafka:
        groupId:
          type: string
          enum: ["some.consumer.service"]
        error-topics:
          addressTemplate: "${groupId}.__.${channel.address}.${suffix}"
          retryTopics: 3
          retry:
            partitions: 1
            replicas: 2
            topicConfiguration:
              retention.ms: 259200000
          dlq:
            partitions: 1
            replicas: 2
            topicConfiguration:
              retention.ms: 2592000000
              cleanup.policy: ["delete"]

Schema

error-topics

Field	Type	Required	Description
addressTemplate	string	Yes	Template for computing topic addresses. See below.
retryTopics	integer	No	Number of retry topics. Creates suffixes retry-0 through retry-{N-1}. Required when retry is present.
retry	object	No	Config for retry topics. Omit to skip provisioning.
dlq	object	No	Config for the DLQ topic. Omit to skip provisioning.

retry and dlq

Field	Type	Required	Description
partitions	integer	No	Number of partitions.
replicas	integer	No	Replication factor.
topicConfiguration	object	No	Same schema as channel binding topicConfiguration.

Environment-specific overrides are out of scope here. The env-server-overrides mechanism from #292 applies the same merge semantics and can be extended to cover retry and dlq if that proposal is adopted.

Address Template

Variable	Resolved value
${groupId}	First value of the groupId enum in the operation binding.
${channel.address}	The address of the referenced channel.
${suffix}	retry-0, retry-1, … retry-{N-1}, dlq.

${groupId} resolution. groupId is a schema object (type: string, enum: [...]), not a plain string. Tooling uses the first enum value. If absent or no enum is defined, resolves to empty string and tooling should emit a warning.

Example — groupId: some.consumer.service, channel address some.domain.someservice.some-event.avro.v0, retryTopics: 3, template ${groupId}.__.${channel.address}.${suffix}:

some.consumer.service.__.some.domain.someservice.some-event.avro.v0.retry-0
some.consumer.service.__.some.domain.someservice.some-event.avro.v0.retry-1
some.consumer.service.__.some.domain.someservice.some-event.avro.v0.retry-2
some.consumer.service.__.some.domain.someservice.some-event.avro.v0.dlq

The .__. separator is suggested to visually distinguish consumer group from original topic. Tooling can split on .__. to recover both parts unambiguously.

Reusable Plans via components

components:
  error-topics:
    retry:
      silver:
        partitions: 1
        replicas: 2
        topicConfiguration:
          retention.ms: 259200000       # 3 days
      gold:
        partitions: 3
        replicas: 3
        topicConfiguration:
          retention.ms: 604800000       # 7 days
    dlq:
      standard:
        partitions: 1
        replicas: 2
        topicConfiguration:
          retention.ms: 2592000000      # 30 days
          cleanup.policy: ["delete"]
      compliance:
        partitions: 1
        replicas: 3
        topicConfiguration:
          retention.ms: 31536000000     # 1 year
          cleanup.policy: ["delete"]

operations:
  onSomethingHappened:
    bindings:
      kafka:
        groupId:
          type: string
          enum: ["some.consumer.service"]
        error-topics:
          addressTemplate: "${groupId}.__.${channel.address}.${suffix}"
          retryTopics: 3
          retry:
            $ref: '#/components/error-topics/retry/silver'
          dlq:
            $ref: '#/components/error-topics/dlq/compliance'

🚧 Breaking changes

No

👀 Have you checked for similar open issues?

• I checked and didn't find a similar issue

🏢 Have you read the Contributing Guidelines?

• I have read the Contributing Guidelines

Are you willing to work on this issue?

Yes I am willing to submit a PR!

[dalelane]

I've got a few initial questions:

1. The proposed resolution approach for groupId seems to assume that every groupId is an enum, which seems odd to me. Why not just leave it at the idea that consumers in a certain group can have a dedicated DLQ topic that is named including their consumer group id? Starting to introduce a requirement that groupIds need to be enums feels excessive.

2. Is the suffix/retryTopics thing really needed? This feels like a very specific approach/implementation, rather than something generic enough to document a wide range of event processing applications. Do we need a new top-level components entry for this? If we dropped the suffix/retryTopics aspect, then you could just have a ref to a channel for retry/DLQ topics.

3. How/where would you describe the contents of the DLQ messages? For example, it's common to include details of the failure/error in the message that goes onto the dead letter topic, so where would that be documented?

[ivangsa]

How/where would you describe the contents of the DLQ messages?

This is interesting, and I didn't think about it... my practical answer would be: "conventionally". I expect the payload and the partitioning key to be exactly the same as the original message, but for the particular error headers it depends on the library: Spring Kafka, Spring Cloud Stream have their convention..
I didn't think about how to add some 'traits' so it doesnt get very vebose

seems to assume that every groupId is an enum

I assume that groupId is a single value, I used an enum bc I copied example from the readme and I didnt know I could use a const..

I asume is a single value bc an AsyncAPI v3 "SHOULD describe the operations an application performs" (the enfasis on an is mine), so for one application and one operation I expect a single value for the groupId. Even if the same app need to read from the same topic more than once, I would expect a different operation and groupId. (The only reason I can think for more than one groupId is if two different version of an app coexists on the same environment, something like piloting a new feature/implementation)

Is the suffix/retryTopics thing really needed? [...] could just have a ref to a channel for retry/DLQ topics

Well, this entire feature is about avoiding "just have a ref to a channel for retry/DLQ" because: that will be very verbose, and even worse is that those retry/dlq are not even the API surface but internal thing of the app..

Let me explain why this came about:

If an app needs lets say 5 retry topics (for non-blocking retry) and we want to provision all infra from the asyncapi.yml we will need to define:

• the original topic and the original operation
• 5 extra topics (sharing the same message/s, headers)
• 10 extra operations for send / receive to the retry topics, so we can grant ACL access to those topics
• one dlq topic + one extra operation

And if instead of one original topic, you have 3, 5... This already out of hands (in verbosity)...

Do we need a new top-level components entry for this?

I think this top-level component solves this verbosity problem not bad..

the suffix

I undestand the part of the addressTemplate is where this feature gets opinionated... If we don't want to specify the name of all retry/dlq topics (and we don't) we need a pattern, and the minimun we need for that pattern is a suffix (but I understand that the content of how suffix is build should be left more open)

Regarding ${groupId} and ${channel.address} I jumped into tooling thinking here, but these are the minimun you need to build a unique topic name pattern..
Build the name arround the "original topic name", and if the topic has more than one consumer... then you also need the "consumer group" otherwise you mix everything up..

---

So these is the rational about this feature.. I know we have some other pending.. this one introduces a "not small" top-level component

[ivangsa]

How/where would you describe the contents of the DLQ messages?
After some thought I'd like to propose adding extraHeaders to the root of error-topics

The name reflects what these headers actually are: headers injected on top of the original message by the error-handling framework (Spring Kafka, Spring Cloud Stream, etc.). Since the field lives inside error-topics, the error context is already implied. extraHeaders is precise without being redundant.

As clarification, the message schema and headers for error topics and dlq are the same as the original topic, this is extra metadata on top of that message

I will update the PR acordingly