Unit 4.3: Event Streaming Integration

Introduction

You are wiring TaskBoard's API service to its background worker. When a user creates, moves, or closes a task, the API must not block on sending an email or a webhook. Instead, the API produces a task-changed event to Kafka and returns immediately, while a separate worker service consumes those events and fires notifications. This decouples the request path from slow side effects and lets you scale the worker independently.

IONOS Event Streams for Apache Kafka runs Apache Kafka 4.0.0, so your existing Kafka client libraries work unchanged. What differs from a self-managed broker is the connection: the data plane requires mutual TLS, not SASL or plaintext, and the broker addresses and certificates come from the cluster you provisioned in Unit 2.5. This unit starts at the client connection, then builds up topic design, reliable delivery, dead-letter handling, and consumer scaling, all against the TaskBoard event flow.

1. Connecting Clients with mTLS

The IONOS Kafka data plane is TLS secured with mutual authentication. Both sides authenticate each other: your client verifies the broker certificate against the cluster's certificate authority, and the broker verifies your client certificate, which the cluster's own certificate authority signs. There is no plaintext or SASL/PLAIN listener. Every producer and consumer presents a client certificate.

You retrieve the credentials, the certificate authority, the client private key, and the client certificate, from the cluster access endpoint. These are the same values the cluster exposes through its broker addresses and access API. Each client certificate is valid for 365 days, so certificate rotation belongs in your operational runbook before the year is out.

1.1 Client Connection Configuration

A self-managed Kafka client config uses security.protocol=SSL with a truststore (for the cluster CA) and a keystore (for the client certificate and key). The IONOS documentation shows the canonical client properties file:

security.protocol=SSL
ssl.truststore.type=PKCS12
ssl.truststore.location=ca-cert.p12
ssl.truststore.password=changeit
ssl.endpoint.identification.algorithm=

ssl.keystore.type=PKCS12
ssl.keystore.location=admin.p12
ssl.keystore.password=adminp12pass

Note ssl.endpoint.identification.algorithm is left empty. Because the cluster uses a private CA rather than publicly signed certificates, hostname verification against the broker addresses is disabled here. The truststore validates the chain instead.

1.2 Producer Connection in Python

The confluent-kafka library accepts the same SSL keys. Point the client at the broker addresses from your Terraform output and supply the CA, certificate, and key as PEM files.

from confluent_kafka import Producer

# Broker addresses come from the ionoscloud_kafka_cluster Terraform output.
conf = {
    "bootstrap.servers": "192.168.1.101:9093,192.168.1.102:9093,192.168.1.103:9093",
    "security.protocol": "ssl",
    "ssl.ca.location": "ca-cert.pem",
    "ssl.certificate.location": "client-cert.pem",
    "ssl.key.location": "client-key.pem",
    "ssl.endpoint.identification.algorithm": "none",
    "client.id": "taskboard-api",
}

producer = Producer(conf)

The cluster runs three brokers across the cluster topology, so list all three in bootstrap.servers for connection resilience. The client only needs one reachable broker to bootstrap, but listing all three avoids a single point of failure during startup.

2. Topic and Partition Design

Topic design is where IONOS-specific constraints first bite. Use one topic per event type rather than a single firehose topic, because consumers subscribe to the events they care about and you can set retention and partitioning per concern. For TaskBoard you create a task-changed topic for the worker, and later a task-changed.dlq dead-letter topic.

2.1 Creating Topics via the API

The Kafka management API is regional and separate from the cloudapi. The host follows the pattern https://kafka.<region>.ionos.com, and unlike the data plane, the management API authenticates with a Bearer token, not mTLS. Create a topic with POST /clusters/{clusterId}/topics:

curl -X POST \
  'https://kafka.de-txl.ionos.com/clusters/e69b22a5-8fee-56b1-b6fb-4a07e4205ead/topics' \
  --header 'Content-Type: application/json' \
  --header 'Authorization: Bearer '"$IONOS_TOKEN" \
  --data '{
    "properties": {
      "name": "task-changed",
      "replicationFactor": 3,
      "numberOfPartitions": 6,
      "logRetention": { "retentionTime": 604800000 }
    }
  }'

The body parameters are name (the only mandatory field), replicationFactor, numberOfPartitions, and the logRetention settings. Read topics back with GET /clusters/{clusterId}/topics and GET /clusters/{clusterId}/topics/{topicId}, which return name, replicationFactor, numberOfPartitions, and logRetention. Deletion is DELETE /clusters/{clusterId}/topics/{topicId} and returns 202 Accepted, since the removal is asynchronous.

2.2 Partition Count and Replication Constraints

Choose the number of partitions as a multiple of 3 (3, 6, 9, 12, and so on) to avoid an uneven distribution of partitions across the three brokers. A partition count that is not a multiple of the broker count leaves one broker carrying more partitions than the others, which skews load.

The recommended replication factor is 3, matching the three-broker topology, but this is an IONOS recommendation, not a system-enforced default; you must set replicationFactor explicitly when creating a topic. With replication factor 3 the cluster keeps three copies of every message, so the topic survives the loss of a broker without data loss. The default topic retention is 604800000 ms (7 days). Setting retentionTime to -1 applies no time limit, in which case retention is bounded only by the cluster storage you provisioned (for example 750 GB on the S size, 1200 GB on M).

The following table maps the topic body parameters to their role when you design a topic.

Partitions matter beyond storage. The partition count is the hard ceiling on consumer parallelism within a consumer group, so size it for peak throughput now, since increasing partitions later changes key-to-partition mapping and disrupts ordering guarantees.

3. Producer Reliability

A fire-and-forget producer drops messages on broker hiccups. For TaskBoard's task events, where a lost event means a missed notification, configure the producer for durability before you optimize for throughput.

3.1 Acks, Idempotence, and Keys

Set acks=all so the producer waits for all in-sync replicas to acknowledge a write before considering it successful. Combined with replication factor 3, this means a message is durable across brokers before your code moves on. Enable the idempotent producer so retries do not create duplicates on the broker side.

from confluent_kafka import Producer
import json

conf = {
    "bootstrap.servers": "192.168.1.101:9093,192.168.1.102:9093,192.168.1.103:9093",
    "security.protocol": "ssl",
    "ssl.ca.location": "ca-cert.pem",
    "ssl.certificate.location": "client-cert.pem",
    "ssl.key.location": "client-key.pem",
    "ssl.endpoint.identification.algorithm": "none",
    "acks": "all",
    "enable.idempotence": True,
    "retries": 5,
    "linger.ms": 20,
    "batch.size": 65536,
}
producer = Producer(conf)

def on_delivery(err, msg):
    if err:
        # Persist for replay; never silently drop.
        print(f"delivery failed: {err}")
    else:
        print(f"delivered to {msg.topic()} [{msg.partition()}] @ {msg.offset()}")

def emit_task_changed(task_id: str, event: dict):
    producer.produce(
        topic="task-changed",
        key=task_id.encode("utf-8"),   # same task -> same partition -> ordered
        value=json.dumps(event).encode("utf-8"),
        on_delivery=on_delivery,
    )
    producer.poll(0)

Keying by task_id routes all events for one task to the same partition, which preserves per-task ordering. Without a key, Kafka distributes records round-robin and a "task closed" event could be processed before "task created".

3.2 Batching and Flushing

linger.ms and batch.size trade a few milliseconds of latency for far higher throughput by batching records per partition. Always flush() before the process exits, or buffered records are lost.

# At shutdown, block until all buffered messages are delivered or fail.
remaining = producer.flush(timeout=10)
if remaining > 0:
    raise RuntimeError(f"{remaining} messages not delivered before shutdown")

4. Consumer Groups, Offsets, and Delivery Semantics

The TaskBoard worker reads task-changed as a consumer group. Every consumer in the same group shares the partitions, so adding worker pods scales throughput up to the partition count. A fourth consumer on a 3-partition topic sits idle.

4.1 At-Least-Once with Manual Commits

For notification delivery, at-least-once is the right default: process the message, then commit the offset. If the worker crashes after processing but before committing, the message is redelivered and the notification may fire twice, which you make safe with an idempotency key. The opposite, committing before processing (at-most-once), risks losing events on a crash.

from confluent_kafka import Consumer, KafkaException
import json

conf = {
    "bootstrap.servers": "192.168.1.101:9093,192.168.1.102:9093,192.168.1.103:9093",
    "security.protocol": "ssl",
    "ssl.ca.location": "ca-cert.pem",
    "ssl.certificate.location": "client-cert.pem",
    "ssl.key.location": "client-key.pem",
    "ssl.endpoint.identification.algorithm": "none",
    "group.id": "taskboard-notifier",
    "enable.auto.commit": False,        # manual commit = control over semantics
    "auto.offset.reset": "earliest",
}
consumer = Consumer(conf)
consumer.subscribe(["task-changed"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        raise KafkaException(msg.error())
    event = json.loads(msg.value())
    send_notification(event)            # do the work first
    consumer.commit(msg)                # then commit only this offset

enable.auto.commit=False is the key choice. Auto-commit advances offsets on a timer regardless of whether processing succeeded, which silently drops messages whose processing failed.

4.2 Consumer Scaling and Rebalancing

The partition count is the parallelism ceiling. To scale the TaskBoard worker on Managed Kubernetes (see Unit 3.2), increase the Deployment replica count up to the partition count, and each new pod joins the taskboard-notifier group and is assigned a subset of partitions through a rebalance. During a rebalance, partitions briefly stop being consumed, so keep processing idempotent and commits frequent to minimize reprocessing.

Size partitions for your peak consumer count up front. You can add partitions later, but doing so reshuffles key-to-partition mapping and breaks ordering for in-flight keys.

5. Dead-Letter Handling and Schema Discipline

Not every message can be processed. A malformed payload or a downstream outage will keep failing on redelivery, and a poison message that you retry forever blocks its entire partition.

5.1 Dead-Letter Topic Pattern

Route messages that exhaust their retries to a dedicated dead-letter topic, task-changed.dlq, then commit the original offset so the main partition keeps flowing. You inspect the DLQ later, fix the cause, and optionally replay.

from confluent_kafka import Producer, Consumer
import json

dlq = Producer(conf_producer)  # same mTLS settings as the main producer

def process_with_dlq(consumer, msg, max_attempts=3):
    attempts = int(dict(msg.headers() or {}).get("x-attempts", b"0") or 0)
    try:
        send_notification(json.loads(msg.value()))
        consumer.commit(msg)
    except Exception as exc:
        if attempts + 1 >= max_attempts:
            dlq.produce(
                "task-changed.dlq",
                key=msg.key(),
                value=msg.value(),
                headers={"x-error": str(exc), "x-attempts": str(attempts + 1)},
            )
            dlq.flush(5)
            consumer.commit(msg)        # advance past the poison message
        else:
            raise                        # let it redeliver for a transient error

Distinguish transient failures (a timeout on the notification service, which deserves a retry) from permanent ones (a malformed event, which goes straight to the DLQ). Blindly retrying permanent failures stalls the partition; blindly dead-lettering transient failures loses recoverable events.

5.2 Event Contracts with JSON Schema

Producers and consumers are deployed independently, so the event payload is a contract between them. Define it with JSON Schema and validate on both sides. Evolve the schema additively: add optional fields, never remove or rename existing ones, so old consumers keep working while new producers ship.

import jsonschema

TASK_CHANGED_V1 = {
    "type": "object",
    "required": ["task_id", "action", "occurred_at"],
    "properties": {
        "task_id": {"type": "string"},
        "action": {"enum": ["created", "moved", "closed"]},
        "occurred_at": {"type": "string", "format": "date-time"},
        "assignee": {"type": "string"},   # added in v1.1, optional = safe
    },
    "additionalProperties": False,
}

def validate_event(event: dict):
    jsonschema.validate(event, TASK_CHANGED_V1)  # raises ValidationError on breach

Carry a schema version in the event or a header so consumers can branch on it during a migration window. This discipline is what lets you change TaskBoard's event shape without a coordinated, lockstep redeploy of producer and worker.

API Reference Quick Card

Key endpoints for the Kafka management API. Host is regional, for example https://kafka.de-txl.ionos.com:

Management API auth: Authorization: Bearer <token> Data plane auth (producers/consumers): mutual TLS with client certificate

Code Lab

Objective: Produce and consume TaskBoard task-changed events on an IONOS Kafka cluster with a consumer group, manual offset commits, and dead-letter routing on failure.

Prerequisites:

• IONOS Cloud account with API token (IONOS_TOKEN)
• A provisioned Kafka cluster from Unit 2.5 with broker addresses available
• Python 3.10+ with confluent-kafka and jsonschema installed
• The cluster CA, client certificate, and client key saved as PEM files

Step 1: Fetch broker addresses and access credentials

CLUSTER_ID="e69b22a5-8fee-56b1-b6fb-4a07e4205ead"
curl -s -X GET \
  "https://kafka.de-txl.ionos.com/clusters/$CLUSTER_ID" \
  -H "Authorization: Bearer $IONOS_TOKEN" | python3 -m json.tool

Expected output:

{
  "id": "e69b22a5-8fee-56b1-b6fb-4a07e4205ead",
  "properties": { "name": "...", "version": "4.0.0", "size": "S",
    "connections": [ { "brokerAddresses": ["192.168.1.101/24", ...] } ] }
}

Step 2: Create the main and dead-letter topics

for T in task-changed task-changed.dlq; do
  curl -s -X POST "https://kafka.de-txl.ionos.com/clusters/$CLUSTER_ID/topics" \
    -H "Content-Type: application/json" -H "Authorization: Bearer $IONOS_TOKEN" \
    --data "{\"properties\":{\"name\":\"$T\",\"replicationFactor\":3,\"numberOfPartitions\":6}}"
done

Expected output:

{"id":"ae085c4c-...","type":"topic","properties":{"name":"task-changed",
 "replicationFactor":3,"numberOfPartitions":6}}

Step 3: Produce a task-changed event

producer.produce("task-changed", key=b"task-42",
    value=b'{"task_id":"task-42","action":"created","occurred_at":"2026-06-05T10:00:00Z"}')
producer.flush(10)
print("produced")

Expected output:

delivered to task-changed [3] @ 0
produced

Step 4: Consume with a consumer group and manual commit

consumer.subscribe(["task-changed"])
msg = consumer.poll(5.0)
print(msg.topic(), msg.partition(), msg.offset(), msg.value())
consumer.commit(msg)

Expected output:

task-changed 3 0 b'{"task_id":"task-42",...}'

Step 5: Trigger a dead-letter on a bad payload

producer.produce("task-changed", key=b"task-99", value=b'{not valid json')
producer.flush(10)
# Consumer's process_with_dlq routes it to task-changed.dlq after max_attempts

Expected output:

delivered to task-changed.dlq [1] @ 0

Step 6: Verify the dead-letter topic received the message

dlq_consumer.subscribe(["task-changed.dlq"])
m = dlq_consumer.poll(5.0)
print(dict(m.headers()), m.value())

Expected output:

{'x-error': b'...', 'x-attempts': b'3'} b'{not valid json'

Validation Checklist:

• [ ] task-changed and task-changed.dlq topics created with replication factor 3
• [ ] Producer delivers with acks=all and idempotence enabled
• [ ] Consumer commits offsets manually after processing
• [ ] A malformed message lands in the dead-letter topic, not an infinite retry loop

Cleanup:

for ID in $(curl -s "https://kafka.de-txl.ionos.com/clusters/$CLUSTER_ID/topics" \
  -H "Authorization: Bearer $IONOS_TOKEN" | python3 -c \
  "import sys,json;[print(t['id']) for t in json.load(sys.stdin)['items']]"); do
  curl -s -X DELETE "https://kafka.de-txl.ionos.com/clusters/$CLUSTER_ID/topics/$ID" \
    -H "Authorization: Bearer $IONOS_TOKEN"
done

Common Pitfalls

1. Using SASL or plaintext instead of mTLS

   • Problem: The client hangs on connect or fails with SSL handshake failed, even though credentials look correct.
   • Why it happens: The IONOS data plane only accepts mutual TLS. There is no SASL/PLAIN or plaintext listener, and the brokers use a private CA, so a default truststore rejects them.
   • Fix: Set security.protocol=ssl, supply the cluster CA as ssl.ca.location, the client certificate and key, and disable hostname verification with ssl.endpoint.identification.algorithm=none.

2. Auto-commit silently dropping failed messages

   • Problem: Some task notifications never fire, but no errors appear in the logs and the consumer lag is zero.
   • Why it happens: With enable.auto.commit=True, offsets advance on a timer regardless of whether send_notification succeeded, so a message that raised an exception is marked as consumed.
   • Fix: Set enable.auto.commit=False and call consumer.commit(msg) only after processing succeeds, routing permanent failures to the dead-letter topic.

3. Partition count not a multiple of the broker count

   • Problem: One broker shows higher CPU and disk load than the other two, and throughput plateaus below expectations.
   • Why it happens: With 3 brokers, a topic created with 4 or 5 partitions distributes unevenly, so one broker hosts an extra partition leader.
   • Fix: Create topics with a partition count that is a multiple of 3 (3, 6, 9, 12) so partitions spread evenly across the three brokers.

Summary

You can now integrate IONOS Event Streams for Apache Kafka into application code as a first-class part of the TaskBoard architecture. You connect producers and consumers over mutual TLS using broker addresses and certificates from your provisioned cluster, create topics through the regional management API with Bearer auth, and design partition and replication settings that respect the three-broker topology. On top of that you have a reliable producer with idempotence and acks=all, a consumer group with manual at-least-once commits, a dead-letter topic for poison messages, and JSON Schema contracts that let producer and consumer evolve independently.

Key Points:

• The Kafka data plane requires mutual TLS with a client certificate; the management API uses Bearer tokens. They are two different auth models on two different endpoints.
• Topic partition counts should be multiples of 3 to spread evenly across the three brokers; the recommended replication factor is 3, but you must set it explicitly since it is not a system-enforced default.
• At-least-once delivery comes from disabling auto-commit and committing offsets only after successful processing; make processing idempotent to tolerate redelivery.
• The partition count is the hard ceiling on consumer parallelism; size partitions for peak load before deploying, because adding them later disrupts key ordering.
• Dead-letter topics isolate poison messages so one bad payload does not block its partition, and JSON Schema enforces the producer-to-consumer contract.

Important Terminology:

• mTLS (mutual TLS): Both client and broker authenticate each other with certificates; the IONOS data plane's only auth method, with client certificates valid for 365 days.
• Consumer group: A set of consumers sharing the work of a topic's partitions, the unit of horizontal scaling for the TaskBoard worker.
• Offset commit: Recording the position a consumer has processed up to; committing after processing yields at-least-once delivery.
• Dead-letter topic (DLQ): A separate topic for messages that fail processing after retries, keeping the main partition unblocked.
• Replication factor: The number of copies of each partition across brokers; 3 on IONOS, matching the three-broker cluster, so the topic survives a broker loss.

Next Steps

Continue Learning: Unit 4.4: AI Model Hub Integration

Related Topics:

• Unit 2.5: Database and Streaming Service Provisioning
• Unit 3.2: Kubernetes Deployment and Operations