-   [](https://platform.exaloop.io/)
-   [Cookbook](https://platform.exaloop.io/cookbook/)
-   Normalize Orders in a Kafka Transaction

# Normalize Orders in a Kafka Transaction

Normalize an Avro order stream and publish the cleaned records with their processed input offsets in a single Kafka transaction. The worker validates business fields, standardizes status casing, and preserves keys and tombstones. Downstream read-committed consumers see only committed output.

**Uses:** Kafka SDK, Data SDK, generated Avro records. **Services:** a development Kafka cluster with transaction support. **Difficulty:** advanced.

This is a Kafka-to-Kafka transaction in one cluster. It does not make an HTTP request, a payment, a database write, or a Parquet file atomic with the offsets.

## Create the Contract and Topics

Use fresh, disposable topics and a dedicated group for this three-record example. It preserves tombstones and message keys. It deliberately uses raw Avro values: every participant must know the same writer schema. These are neither Avro container files nor schema-registry-framed messages.

```json title="order.avsc"
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "status", "type": "string"},
    {"name": "amount_cents", "type": "long"}
  ]
}
```

```sh title="Terminal"
exaloop avrogen order.avsc order.codon
export KAFKA_BROKERS=localhost:9092
kafka-topics.sh --bootstrap-server "$KAFKA_BROKERS" --create --topic cookbook-orders --partitions 1 --replication-factor 1
kafka-topics.sh --bootstrap-server "$KAFKA_BROKERS" --create --topic cookbook-normalized --partitions 1 --replication-factor 1
```

`kafka-topics.sh` is Kafka's administration tool, not supplied by the Codon SDK. Use your cluster's equivalent tool if needed. A single-broker development cluster must also have transaction-state replication settings compatible with one broker. Production replication and availability settings should follow your cluster policy, not this fixture.

## Seed, Transform, and Verify

The one executable has three modes. `seed` writes two orders and a tombstone; `transform` processes exactly three inputs; `verify` reads the output partition from the beginning without committing group offsets. Each read fails after a 60-second deadline instead of waiting forever.

```python title="orders.codon"
import os
import sys
import time
import exaloop.streaming.kafka as kafka
from order import Order

brokers = os.environ.get("KAFKA_BROKERS", "localhost:9092")
source = "cookbook-orders"
destination = "cookbook-normalized"

def receive(consumer, limit: int, deadline: float):
    while time.monotonic() < deadline:
        batch = consumer.poll_batch(limit, 1.0)
        if batch:
            return batch
    raise ValueError("input deadline exceeded; check topics, offsets, and broker")

def seed(brokers: str, source: str):
    payloads = {
        b"order-1": Order(1, "paid", 12500).to_bytes(),
        b"order-2": Order(2, "Refunded", 7800).to_bytes(),
        b"order-3": None,
    }
    with kafka.Producer(brokers) as producer:
        for key, payload in payloads.items():
            producer.produce(source, payload, key=key)
        reports = producer.flush(30.0)
        for report in reports:
            report.raise_for_error()
        assert len(reports) == 3
    print("seeded 3")

def transform(brokers: str, source: str, destination: str):
    processed = 0
    deadline = time.monotonic() + 60.0
    with kafka.Consumer(brokers, "cookbook-normalizers",
                        {"isolation.level": "read_committed"}) as consumer:
        consumer.subscribe([source])
        with kafka.Producer(brokers, {"transactional.id": "cookbook-normalizer-0"}) as producer:
            producer.init_transactions()
            decoder = Order.decoder()
            while processed < 3:
                batch = receive(consumer, 3 - processed, deadline)
                producer.begin_transaction()
                for message in batch:
                    payload = None
                    if message.value is not None:
                        order = decoder.decode(message.value)
                        status = order.status.upper()
                        if order.id <= 0 or order.amount_cents < 0 or status not in ("PAID", "REFUNDED"):
                            raise ValueError("invalid order")
                        payload = Order(order.id, status, order.amount_cents).to_bytes()
                    producer.produce(destination, payload, key=message.key, headers=message.headers)
                producer.send_offsets_to_transaction(consumer, batch)
                for report in producer.commit_transaction():
                    report.raise_for_error()
                processed += len(batch)
    print("committed", processed)

def verify(brokers: str, destination: str):
    records: dict[int, str] = {}
    tombstones = 0
    received = 0
    deadline = time.monotonic() + 60.0
    with kafka.Consumer(brokers, "cookbook-verification",
                        {"isolation.level": "read_committed"}) as consumer:
        consumer.assign([kafka.TopicPartition(destination, 0, 0)])
        while received < 3:
            for message in receive(consumer, 3 - received, deadline):
                if message.value is None:
                    assert message.key == b"order-3"
                    tombstones += 1
                else:
                    order = Order.from_bytes(message.value)
                    assert message.key == ("order-" + str(order.id)).encode()
                    assert order.id not in records
                    records[order.id] = order.status
                received += 1
    assert records == {1: "PAID", 2: "REFUNDED"} and tombstones == 1
    print("verified 2 orders and 1 tombstone")

mode = sys.argv[1] if len(sys.argv) > 1 else "transform"
if mode == "seed":
    seed(brokers, source)
elif mode == "transform":
    transform(brokers, source, destination)
elif mode == "verify":
    verify(brokers, destination)
else:
    raise ValueError("expected seed, transform, or verify")
```

```sh title="Terminal"
exaloop --with data --with kafka build -release -o orders orders.codon
./orders seed
./orders transform
./orders verify
```

```text title="Output"
seeded 3
committed 3
verified 2 orders and 1 tombstone
```

Do not repeat `seed` on these same topics when checking this exact result: it creates new input records. A completed `transform` resumes after its committed offsets and will time out without new input. Reset only disposable fixtures with your Kafka administration tools; do not reset a production consumer group's offsets to make a demo rerun.

## Recovery and Deployment

The bounded example stops on a decode, delivery, transaction, or queue-capacity failure. Leaving the producer scope attempts to abort an active transaction. A long-running service needs an explicit recovery loop using [`KafkaError` flags](https://platform.exaloop.io/streaming/kafka/#transactions): retry a retriable transaction operation; abort and replay after an abort-required failure; replace a producer after a fatal failure. A timed-out commit may still be completing, so retry that commit rather than starting a new transaction. An abort does not rewind the consumer.

Use a stable transactional ID per logical worker, with different IDs for concurrently active workers. A replacement using the same ID fences the old instance. Never add a separate `consumer.commit()` to this transaction path. Keep batch work within the group's poll and transaction deadlines, and handle backpressure by polling delivery reports. The three-record fixture is not a throughput configuration.

Malformed input intentionally blocks progress rather than disappearing behind a committed offset. A production policy can publish a controlled quarantine record in the same transaction, but must define payload retention, access, and replay. For evolving registry-backed contracts, replace the raw decoder and encoder with [registry codecs](https://platform.exaloop.io/streaming/kafka/#use-a-schema-registry); do not strip the envelope or assume the latest schema was used to write every message.

Both local connections above use plaintext. Configure [SASL and verified TLS](https://platform.exaloop.io/streaming/kafka/#security-and-deployment) before connecting to a production cluster. Package both SDK runtimes, and verify the destination's broker credentials, CA certificates, and topic permissions.

Source: https://platform.exaloop.io/cookbook/transactional-events/