# Kafka usage examples

Kafka topics - consuming messages through channels and publishing with self.out.kafka.

## Consuming messages from topics {#consuming-messages-from-topics}

No programming is needed to consume messages from Kafka topics. Create a Kafka channel in Dashboard, point it to a topic, and a service of your choice will be invoked for each message received.

[Video: New Kafka channel](https://zatosource-production.b-cdn.net/docs/gfx/channels/kafka-create.webm?v=1790934308)

Dashboard menu: Connections > Message queues > Kafka > Channels

The message payload will be available in self.request.input.

```python
from zato.server.service import Service

class MyService(Service):

    def handle(self):

        # The incoming Kafka message payload
        data = self.request.input

        # Log what was received
        self.logger.info('Kafka message: %s', data)
```

## Publishing messages to topics {#publishing-messages-to-topics}

Create a Kafka outgoing connection in Dashboard and use self.out.kafka to publish messages.

The connection is looked up by name, and .send() accepts strings, bytes, dicts, lists or any other JSON-serializable object.

```python
from zato.server.service import Service

class MyService(Service):

    def handle(self):

        # Get a connection by name
        conn = self.out.kafka['my-publisher']

        # Send a string
        conn.send('Hello from Zato')
```

## Sending structured data {#sending-structured-data}

When you pass a dict or list, it is automatically serialized to JSON before publishing.

```python
from zato.server.service import Service

class MyService(Service):

    def handle(self):
        conn = self.out.kafka['my-publisher']

        conn.send({
            'event': 'order.created',
            'order_id': '12345',
            'customer': 'Acme Corp',
        })
```

## Processing and forwarding {#processing-and-forwarding}

A common pattern is to consume a message from one topic, process it and publish the result to another.

```python
from zato.server.service import Service

class MyService(Service):

    def handle(self):

        # Incoming message from a Kafka channel
        data = self.request.input
        self.logger.info('Processing: %s', data)

        # Forward to another topic through a different outgoing connection
        conn = self.out.kafka['processed-events']
        conn.send(data)
```

## Authentication {#authentication}

Brokers that require SASL authentication are configured in two steps, on both channels and outgoing connections - the connection names a SASL mechanism and a security definition the mechanism takes its credentials from. There is no Kafka-specific credential store, the same security definitions that REST connections use serve here.

| Mechanism | Security definition | What is sent |
| --- | --- | --- |
| `PLAIN` | Basic Auth | The definition's username and password |
| `SCRAM-SHA-256` | Basic Auth | The definition's username and password |
| `SCRAM-SHA-512` | Basic Auth | The definition's username and password |
| `OAUTHBEARER` | Bearer token | A token obtained from the definition's token endpoint with its client ID (the username), client secret (the password) and scopes |

A mechanism is given together with a security definition or not at all - one without the other is rejected when the connection is saved. When the definition's password or client secret changes later on, every Kafka connection using it switches to the new credentials automatically.

With TLS on, the connection uses the `sasl_ssl` protocol, without it `sasl_plaintext`.

A username or password that starts with `$` is read as the name of an environment variable. When the literal value is needed, double the sign - Azure Event Hubs takes `$ConnectionString` as the `PLAIN` username, which is entered as `$$ConnectionString`.

In the Dashboard, both fields are under the Authentication link of a connection's form. In [enmasse](https://zato.io/docs/admin/enmasse-reference.html#kafka-channels), they are the `security` and `sasl_mechanism` keys.

```yaml
security:
  - name: kafka.scram
    type: basic_auth
    username: zato
    password: Zato_Enmasse_Env.KafkaPassword

channel_kafka:
  - name: incoming-events
    address: broker1:9093
    topic: events
    group_id: zato-consumers
    service: kafka.process-event
    security: kafka.scram
    sasl_mechanism: SCRAM-SHA-512
    ssl: true
    ssl_ca_file: /path/to/ca.pem
```

## Learn more {#learn-more}

- [Development documentation](https://zato.io/docs/dev/) - The complete reference for writing services
- [Requests and responses](https://zato.io/docs/dev/request-response/) - What a service receives, what it returns and how both are structured
- [Integration examples](https://zato.io/docs/dev/examples/) - Ready-made code for the systems you are likely to connect to
- [IDE and debugging](https://zato.io/docs/dev/ide/) - Write services in the Dashboard or in your own editor
- [Data models](https://zato.io/docs/dev/model/) - Declare inputs and outputs and have them validated for you
- [In-depth API tutorial](https://zato.io/tutorials/main/01.html) - The full platform tutorial, from installation to production patterns