Kafka bindings

Found an error? Have a suggestion?Edit this page on GitHub

Introduction

You learned how to manage schemas with a schema registry in the previous tutorial. This tutorial teaches you how Kafka bindings function by defining Kafka messages and expanding your AsyncAPI document with protocol-specific details.

Background context

Bindings are essential for event-driven applications because they provide protocol-specific details, abstracting the complexities of message handling from your application's core logic. They enhance the API's clarity and usability by offering setup options and context for different protocols. Bindings include the topics your application reads from or writes to, message formatting, and rules for interacting with multiple data or messages.

In an AsyncAPI document, bindings can be added to various sections like servers, channels, or messages. They contain protocol-specific details unique to each protocol. Binding definitions let you specify functionalities specific to the protocol, which are not covered by AsyncAPI's core features.

You can configure several objects using Kafka bindings. However, for the scope of this tutorial, you will focus on four levels of bindings: server bindings, operations binding, channel bindings, and message bindings.

Using the code snippets from the previous tutorial, where you learned how to manage Avro schemas using a centralized schema registry that enables you to share schemas across multiple applications, you will add configurations for server, operations, channel, and message bindings.

Below, you can find the updated schema reference file you'll use for this tutorial.

1asyncapi: 3.0.0
2info:
3  title: User Signup API
4  version: 1.0.0
5  description: The API notifies you whenever a new user signs up in the application.
6servers:
7  kafkaServer:
8    host: test.mykafkacluster.org:8092
9    description: Kafka Server
10    protocol: kafka
11operations:
12  onUserSignedUp:
13    action: receive
14    channel:
15      $ref: '#/channels/userSignedUp'
16channels:
17  userSignedUp:
18    description: This channel contains a message per each user who signs up in our application.
19    address: user_signedup
20    messages:
21      userSignedUp:
22        $ref: '#/components/messages/userSignedUp'
23components:
24  messages:
25    userSignedUp:
26      payload:
27        schemaFormat: 'application/vnd.apache.avro+json;version=1.9.0'
28        schema:
29          $ref: http://localhost:8080/apis/registry/v2/groups/my-group/artifacts/UserSignedUp

Add server bindings

Server bindings provide protocol-specific configuration details for connecting and interacting with a server.

Server bindings allow you to specify a schemaRegistryUrl, which provides an API URL for a given server where a schema registry was used. A schema registry is a repository for managing and validating messages' schemas. To learn more about schema registry, read the message validation guide for schema registry.

schemaRegistryVendor is used optionally to refer to vendors or platforms that provide the schema registry service, in this case, Apicurio Registry. Learn about other fields you can configure under server bindings.

1servers:
2  kafkaServer:
3    host: test.mykafkacluster.org:8092
4    description: Kafka Server
5    protocol: kafka
6    bindings:
7      kafka:
8        schemaRegistryUrl: 'http://localhost:8080/apis/registry/'
9        schemaRegistryVendor: 'apicurio'
10        bindingVersion: '0.5.0'

Important: bindingVersion is the field version of a binding. It specifies the version of the binding specification that is used to describe how an API interacts with Kafka. The bindingVersion field is an optional field that is available for all bindings.

Add operation bindings

Operation bindings object contains information about the operation representation in Kafka (eg. the way to consume messages).

The operation binding object provides a structured way to describe how a particular operation (publish, subscribe) should behave on a Kafka topic. The groupid, for example, is the Id of the consumer group, while the cliendID is the Id of the consumer within a consumer group.

These configurations are vital for distributed message consumption and load balancing among consumers. Learn more about other fields you can configure under operations binding.

1operations:
2  onUserSignedUp:
3    action: receive
4    channel:
5      $ref: '#/channels/userSignedUp'
6    bindings:
7      kafka:
8        bindingVersion: '0.5.0'
9        groupId:
10          type: string
11          enum: ['myGroupId']
12        clientId:
13          type: string
14          enum: ['myClientId']

Add channel bindings

Channel bindings provide protocol-specific information for a particular channel.

These configurations may include information how the Kafka topic has been configured. The Channel Binding Object is part of AsyncAPI's wider bindings architecture, which specifies how the API interacts with the messaging system — in this case, Kafka.

In Kafka, you can specify a given topic's number of partitions or replicas therefore, enabling parallel processing of data or consumers. Learn more about other fields that you can configure under channel bindings.

1channels:
2  userSignedUp:
3    description: This channel contains a message per each user who signs up in our application.
4    address: user_signedup
5    messages:
6      userSignedUp:
7        $ref: '#/components/messages/userSignedUp'
8    bindings:
9      kafka:
10        bindingVersion: '0.5.0'
11        partitions: 10
12        replicas: 2
13        topicConfiguration:
14          cleanup.policy: ["delete", "compact"]
15          retention.ms: 604800000
16          retention.bytes: 1000000000
17          delete.retention.ms: 86400000
18          max.message.bytes: 1048588

Add message bindings

Message bindings provide protocol-specific information for a specific message. For Kafka topics, this can include how message keys are used, and details about how serialized message data has been encoded.

For example, the schemaIdLocation field, if specified is used to indicate where the schema identifier (ID) for the message payload's schema is located. It is useful for message serialization and deserialization, enabling consumers to understand how to interpret the message payload.

Learn more about other fields that you can configure under message bindings

1components:
2  messages:
3    userSignedUp:
4      bindings:
5        kafka:
6            key:
7              type: string
8              enum: ['myKey']
9            schemaIdLocation: 'payload'
10            schemaIdPayloadEncoding: 'apicurio-new'
11            schemaLookupStrategy: 'TopicIdStrategy'
12            bindingVersion: '0.5.0'
13      payload:
14        schemaFormat: 'application/vnd.apache.avro+json;version=1.9.0'
15        schema:
16          $ref: http://localhost:8080/apis/registry/v2/groups/my-group/artifacts/UserSignedUp

Congratulations, you've completed the tutorial! Putting these blocks together gives you an AsyncAPI document all ready to go.

1asyncapi: 3.0.0
2info:
3  title: User Signup API
4  version: 1.0.0
5  description: The API notifies you whenever a new user signs up in the application.
6servers:
7  kafkaServer:
8    host: test.mykafkacluster.org:8092
9    description: Kafka Server
10    protocol: kafka
11    bindings:
12      kafka:
13        schemaRegistryUrl: 'http://localhost:8080/apis/registry/'
14        schemaRegistryVendor: 'apicurio'
15        bindingVersion: '0.5.0'
16operations:
17  onUserSignedUp:
18    action: receive
19    channel:
20      $ref: '#/channels/userSignedUp'
21    bindings:
22      kafka:
23        bindingVersion: '0.5.0'
24        groupId:
25          type: string
26          enum: ['myGroupId']
27        clientId:
28          type: string
29          enum: ['myClientId']
30channels:
31  userSignedUp:
32    description: This channel contains a message per each user who signs up in our application.
33    address: user_signedup
34    messages:
35      userSignedUp:
36        $ref: '#/components/messages/userSignedUp'
37    bindings:
38      kafka:
39        bindingVersion: '0.5.0'
40        partitions: 10
41        replicas: 2
42        topicConfiguration:
43          cleanup.policy: ["delete", "compact"]
44          retention.ms: 604800000
45          retention.bytes: 1000000000
46          delete.retention.ms: 86400000
47          max.message.bytes: 1048588
48components:
49  messages:
50    userSignedUp:
51      bindings:
52        kafka:
53          bindingVersion: '0.5.0'
54          key:
55            type: string
56            enum: ['myKey']
57          schemaIdLocation: 'payload'
58          schemaIdPayloadEncoding: 'apicurio-new'
59          schemaLookupStrategy: 'TopicIdStrategy'
60      payload:
61        schemaFormat: 'application/vnd.apache.avro+json;version=1.9.0'
62        schema:
63          $ref: http://localhost:8080/apis/registry/v2/groups/my-group/artifacts/UserSignedUp          

Summary

In this tutorial, you learned how to configure server, operation, message, and channel bindings. You also learned that bindings are essential when integrating Kafka with different systems, platforms, or protocols — especially in API specifications like AsyncAPI.

Next steps

Now that you have completed this tutorial, you can learn more about other Kakfa bindings or protocol-specific bindings.

Was this helpful?
Help us improve the docs by adding your contribution.
OR
Github:AsyncAPICreate Issue on GitHub