Kafka usage examples

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.

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

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

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.

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

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

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

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

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

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.

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, they are the security and sasl_mechanism keys.

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