JSON to Asyncapi Definition Converter

JSON to AsyncAPI 3: Kafka Bindings, Channel Operations, Schema Registry, and Message Correlation

AsyncAPI 3.0 (2023) restructured the spec significantly from 2.x: channels no longer contain operations directly — instead, operations reference channels and specify action: send or action: receive. This matters because a Kafka topic is the same channel whether you're a producer or consumer, but the direction is different for each service. This guide covers the AsyncAPI 3.0 structure, Kafka-specific bindings (consumer group, partition, offset reset), message headers for correlation and tracing, schema registry integration, and the AsyncAPI Generator for producing documentation and service stubs.

Live Example: AsyncAPI 3.0 for a Kafka Event System

# Input JSON event payloads
# OrderCreated event:
{ "order_id": "ord_abc", "customer_id": "cust_001", "total_cents": 4998, "created_at": "2024-03-15T10:00:00Z" }
# PaymentProcessed event:
{ "payment_id": "pay_xyz", "order_id": "ord_abc", "status": "SUCCESS", "processed_at": "2024-03-15T10:00:05Z" }

# Generated AsyncAPI 3.0 definition
asyncapi: 3.0.0
info:
  title:   Order Processing System
  version: 1.0.0
  description: |
    Event-driven order processing pipeline.
    Producers publish to order.created and payment.processed.
    Downstream services subscribe and react.

defaultContentType: application/json

servers:
  production:
    host:     kafka.internal:9092
    protocol: kafka
    security:
      - saslScram: []
  development:
    host:     localhost:9092
    protocol: kafka

channels:
  order-created:
    address:  orders.created.v1       # Kafka topic name
    messages:
      OrderCreatedMessage:
        $ref: '#/components/messages/OrderCreated'
    bindings:
      kafka:
        topic:             orders.created.v1
        partitions:        12
        replicas:          3
        topicConfiguration:
          retentionMs:     604800000  # 7 days
          cleanupPolicy:   [delete]

  payment-processed:
    address:  payments.processed.v1
    messages:
      PaymentProcessedMessage:
        $ref: '#/components/messages/PaymentProcessed'

operations:
  # Order Service: publishes OrderCreated
  publishOrderCreated:
    action:   send
    channel:  { $ref: '#/channels/order-created' }
    messages:
      - $ref: '#/channels/order-created/messages/OrderCreatedMessage'

  # Notification Service: consumes OrderCreated
  receiveOrderCreated:
    action:   receive
    channel:  { $ref: '#/channels/order-created' }
    bindings:
      kafka:
        groupId:          { type: string, enum: [notification-service] }
        clientId:         { type: string }
        autoOffsetReset:  earliest
    messages:
      - $ref: '#/channels/order-created/messages/OrderCreatedMessage'

  # Payment Service: publishes PaymentProcessed
  publishPaymentProcessed:
    action:  send
    channel: { $ref: '#/channels/payment-processed' }

Message Headers for Correlation and Tracing

# Message headers carry metadata separate from the business payload
components:
  messages:
    OrderCreated:
      name:    OrderCreated
      title:   Order Created Event
      summary: Fired when a new order is placed
      headers:
        type: object
        required: [correlationId, traceId, eventType, version]
        properties:
          correlationId:
            type:        string
            format:      uuid
            description: Ties all events from one user action together
          traceId:
            type:        string
            description: OpenTelemetry trace ID for distributed tracing
          eventType:
            type:        string
            const:       order.created
          version:
            type:        string
            pattern:     '^\d+\.\d+$'
            example:     "1.0"
          publishedAt:
            type:        string
            format:      date-time
      payload:
        type: object
        required: [order_id, customer_id, total_cents, created_at]
        properties:
          order_id:    { type: string, pattern: '^ord_[a-z0-9]+$' }
          customer_id: { type: string }
          total_cents: { type: integer, minimum: 0 }
          created_at:  { type: string, format: date-time }
          items:
            type:  array
            items:
              type: object
              required: [sku, qty, price_cents]
              properties:
                sku:         { type: string }
                qty:         { type: integer, minimum: 1 }
                price_cents: { type: integer, minimum: 0 }
      correlationId:
        description: Used to trace the event through the pipeline
        location:    $message.header#/correlationId

    PaymentProcessed:
      payload:
        type: object
        required: [payment_id, order_id, status, processed_at]
        properties:
          payment_id:   { type: string }
          order_id:     { type: string }
          status:
            type: string
            enum: [SUCCESS, FAILED, REFUNDED]
          processed_at: { type: string, format: date-time }
          failure_code:
            type: ['string', 'null']
            description: Set when status is FAILED

AMQP (RabbitMQ) Bindings

# AsyncAPI supports multiple protocols via bindings
channels:
  order-notifications:
    address: order.notifications
    bindings:
      amqp:
        is:       routingKey
        exchange:
          name:       order-events
          type:       topic
          durable:    true
          autoDelete: false
    messages:
      OrderNotification:
        $ref: '#/components/messages/OrderCreated'

servers:
  rabbitmq:
    host:     rabbitmq.internal:5672
    protocol: amqp
    bindings:
      amqp:
        exchange: order-events
        vhost:    /production

# WebSocket bindings
servers:
  websocket:
    host:     wss://realtime.example.com
    protocol: wss

channels:
  live-prices:
    address: /prices
    bindings:
      ws:
        bindingVersion: 0.1.0

AsyncAPI Generator: Documentation and Code Stubs

// AsyncAPI Generator produces HTML docs, server stubs, and clients
// npm install -g @asyncapi/generator

// Generate HTML documentation
// ag asyncapi.yaml @asyncapi/html-template -o ./docs

// Generate Node.js service stub (handles Kafka subscribe boilerplate)
// ag asyncapi.yaml @asyncapi/nodejs-template -o ./src/generated

// Generate Python stub
// ag asyncapi.yaml @asyncapi/python-pydantic-template -o ./src

// Programmatic generation in CI:
import { Generator } from '@asyncapi/generator';
import path from 'path';

const generator = new Generator(
  '@asyncapi/html-template',
  path.resolve('./docs'),
  { templateParams: { singleFile: true } }
);

await generator.generateFromFile('./asyncapi.yaml');

// Validate spec before generating
import { Parser } from '@asyncapi/parser';

const parser = new Parser();
const { document, diagnostics } = await parser.parse(specYaml);

if (diagnostics.length > 0) {
  const errors = diagnostics.filter(d => d.severity === 0);
  if (errors.length > 0) {
    console.error('AsyncAPI spec errors:', errors);
    process.exit(1);
  }
}

Best Practices for Production

• Use AsyncAPI 3.0's action: send/receive model: In 2.x, channels had subscribe and publish which were confusing (from whose perspective?). AsyncAPI 3.0 uses action: send (this service produces) and action: receive (this service consumes), which is unambiguous. Operations can reference the same channel with different actions for different services.
• Always document message headers separately from the payload: Correlation IDs, trace IDs, and schema version in headers are just as important as the event payload. Without header documentation, consumers don't know what correlation metadata is available. The correlationId.location field tells AsyncAPI where to find the correlation ID for end-to-end message tracing.
• Include Kafka topic configuration in bindings: Partition count, replication factor, and retention policy are operational parameters that consumers need to understand. Documenting them in bindings means your AsyncAPI spec is also the record of your Kafka topic configuration, not just the message shape.
• Reference schemas from your Avro Schema Registry via $ref or schemaFormat: For Kafka+Avro systems, use schemaFormat: 'application/vnd.apache.avro;version=1.9.0' in the message schema field and reference the AVSC file. This connects your AsyncAPI documentation directly to the authoritative schema, preventing documentation drift.

FAQ

Q: When should I use AsyncAPI instead of OpenAPI?
A: Use AsyncAPI for event-driven, publish-subscribe, or streaming APIs: Kafka topics, RabbitMQ queues, WebSocket connections, MQTT channels, and Server-Sent Events. Use OpenAPI for request-response REST APIs. A system often needs both — REST for synchronous operations, Kafka events for async side effects — and both specs can coexist in the same repository.

Q: Can I share JSON Schema components between AsyncAPI and OpenAPI specs?
A: Yes. Define shared schemas in a separate file (schemas/product.yaml) and reference them from both specs with relative $ref paths. Both OpenAPI 3.1 and AsyncAPI 3.0 use JSON Schema 2020-12, so schemas written for one are compatible with the other. This is the canonical way to share message types between REST API responses and event payloads.

Q: How do I handle schema evolution for Kafka messages?
A: Maintain backward compatibility: add new optional fields, never remove existing fields, never change field types. Version your topic names (orders.created.v1, orders.created.v2) when you need breaking changes. Document the compatibility level in the AsyncAPI bindings and enforce it via Confluent Schema Registry's compatibility check in CI before deploying schema changes.