Consumer

What is Consumer

The Consumer is a program that continuously pulls records from a stream system and processes the data, either sequentially or in parallel. Typically, we require a ‘pointer’ that tells the consumer where to start pulling the data. In Kafka, it is referred to as an ‘offset’; in AWS Kinesis Stream, it is known as a ‘shard iterator’; and in Pulsar, it is called a ‘message id’.

In the previous document, we introduced the concept of Checkpoint. A consumer program essentially leverages the checkpoint, updating the processing status before and after executing processing logic, and handling errors appropriately. It also persists the checkpoint data to the storage backend every time it changes.

Who Implements What

• get_records() and new() — Plugin/backend developers implement these to pull records from a specific streaming backend (e.g. Kinesis get_records, Kafka poll).

• process_record(record) — End users must implement this with their business logic. Raise an exception to indicate failure; the framework handles retries automatically.

• process_failed_record(record) — End users may override this to send failed records to a DLQ. Default is no-op.

• process_batch(), run() — End users call these to start consuming. Already implemented in BaseConsumer.

What is Dead-Letter-Queue (DLQ)

Some records may still fail after multiple retries. Typically, we aim to ensure smooth data processing without blocking it. In business-critical applications, it’s common practice to route failed data to a dedicated location, often a message queue or another stream system. This allows for debugging and later reprocessing.

In certain use cases, it’s critical to process records strictly in order. If a preceding processing attempt fails, we must stop from processing subsequent records. In such scenarios, we should halt processing and trigger a notification for immediate investigation. In any case, a Dead-Letter Queue (DLQ) serves as an additional fault-tolerant layer for business-critical use cases.

Simple Consumer Example

Below is the sample usage of SimpleConsumer, a simple consumer that read data from the output of SimpleProducer.

simple_consumer.py

# -*- coding: utf-8 -*-

import random
import time
import shutil
import dataclasses
from pathlib import Path

from unistream.api import (
    DataClassRecord,
    SimpleCheckpoint,
    SimpleConsumer,
)


def rand_value() -> int:
    return random.randint(1, 100)


@dataclasses.dataclass(frozen=True)
class MyRecord(DataClassRecord):
    value: int = dataclasses.field(default_factory=rand_value)


class RandomError(Exception):
    pass


@dataclasses.dataclass
class MyConsumer(SimpleConsumer):
    path_target: Path = dataclasses.field(init=False)

    def process_record(self, record: MyRecord) -> str:
        s = record.serialize()
        if random.randint(1, 100) <= 50:
            print(f"❌ {s}")
            raise RandomError(f"random error at record_id = {record.id}")
        else:
            with self.path_target.open("a") as f:
                f.write(f"{s}\n")
            print(f"✅ {s}")
        return s

    def process_failed_record(self, record: MyRecord) -> str:
        s = record.serialize()
        if random.randint(1, 100) <= 0:
            print(f"❌ DLQ:{s}")
            raise RandomError(f"{s}")
        else:
            with self.path_dlq.open("a") as f:
                f.write(f"{s}\n")
            print(f"✅ DLQ: {s}")
        return s


dir_here = Path(__file__).absolute().parent
dir_demo = dir_here.joinpath("simple_consumer_demo")
shutil.rmtree(dir_demo, ignore_errors=True)
dir_demo.mkdir(exist_ok=True)

consumer_id = "simple_consumer_1"
path_checkpoint = dir_demo.joinpath(f"{consumer_id}.checkpoint.json")
path_records = dir_demo.joinpath(f"{consumer_id}.records.json")
path_target = dir_demo.joinpath(f"{consumer_id}.target.txt")
path_dlq = dir_demo.joinpath(f"{consumer_id}.dlq.txt")

checkpoint = SimpleCheckpoint.load(
    checkpoint_file=str(path_checkpoint),
    records_file=str(path_records),
)

consumer = MyConsumer.new(
    record_class=MyRecord,
    checkpoint=checkpoint,
    path_source=dir_here.joinpath(
        "simple_producer_demo", "simple_producer_history.log"
    ),
    path_dlq=path_dlq,
    limit=3,
    delay=1,
)
consumer.path_target = path_target


# --- method 1 ---
# consumer.run()

# --- method 2 ---
def run():
    i = 0
    while 1:
        i += 1
        print(f"--- {i} th pull ---")
        consumer.process_batch()
        if consumer.delay:
            time.sleep(consumer.delay)

run()

Vendor-Specific Consumers

Note

Vendor-specific consumers (AWS Kinesis Stream, Apache Kafka, etc.) are released as separate plugin packages — for example unistream-aws-kinesis, unistream-kafka. Each plugin ships its own concrete Consumer and (when relevant) DLQ variants. Refer to those plugins for installation and usage examples.