Producer

What is Producer

The Producer is a program that continuously generates records and sends them to a target system. To optimize performance, it utilize a buffer to send data in batches. Regardless of whether it’s the producer program or the buffer program that encounters a failure, we must have the capability to restart the program and retry the operation without data loss. This retry process should be executed gracefully, incorporating an exponential backoff strategy. Upon successful completion, we can remove the data from the buffer to save space. This action is referred to as “commit”.

Error Handling

What exponential backoff mean is that, we wait longer and longer between each retry and stop retrying at certain number of failure. The wait internal and max retry count strategy is called a “Schedule”.

When the max retry count is reached, we have two options:

1. skip: persist the context data of this batch for debug or future retry, and skip to the next batch.

2. raise: raise error immediately and stop the program.

SimpleProducer Example

Below is the sample usage of SimpleProducer, a simple producer that send data to a target file on your local machine in append-only mode. This producer is for demo and for testing purpose.

simple_producer.py

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

import typing as T
import time
import random
import shutil
import dataclasses
from pathlib import Path

from unistream.api import (
    SendError,
    DataClassRecord,
    FileBuffer,
    RetryConfig,
    SimpleProducer,
)


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


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


@dataclasses.dataclass
class MyProducer(SimpleProducer):
    def send(self, records: T.List[MyRecord]):
        if random.randint(1, 100) <= 50:
            raise SendError("randomly failed due to send error")
        super().send(records)


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

path_log = dir_demo / "simple_producer_buffer.log"
path_client_target = dir_demo / "simple_producer_history.log"


def make_producer() -> MyProducer:
    producer = MyProducer.new(
        buffer=FileBuffer.new(
            record_class=MyRecord,
            path_wal=path_log,
            max_records=3,
        ),
        retry_config=RetryConfig(
            exp_backoff=[1, 2, 4],
        ),
        path_sink=path_client_target,
    )
    return producer


producer = make_producer()

# --- test 1 ---
n = 15
for i in range(1, 1 + n):
    time.sleep(1)
    # The producer program can be terminated with a 30% chance.
    # we create a new producer object to simulate that.
    if random.randint(1, 100) <= 30:
        producer = make_producer()
    producer.put(DataClassRecord(id=str(i)), verbose=True)

# --- test 2 ---
# producer.retry_config.exp_backoff = [0.1, 0.2, 0.4]
# n = 1000
# for i in range(1, 1 + n):
#     time.sleep(0.001)
#     producer.put(DataClassRecord(id=str(i)), verbose=True)

simple_producer.py Output

+----- ⏱ 📤 Start 'put record' -------------------------------------------------+
📤
📤 record = {"id": "1", "create_at": "2024-01-07T07:31:41.482432+00:00"}
📤 🚫 we should not emit
📤
+----- ⏰ ✅ 📤 End 'put record', elapsed = 0.00 sec -----------------------------+
+----- ⏱ 📤 Start 'put record' -------------------------------------------------+
📤
📤 record = {"id": "2", "create_at": "2024-01-07T07:31:42.486702+00:00"}
📤 🚫 we should not emit
📤
+----- ⏰ ✅ 📤 End 'put record', elapsed = 0.00 sec -----------------------------+
+----- ⏱ 📤 Start 'put record' -------------------------------------------------+
📤
📤 record = {"id": "3", "create_at": "2024-01-07T07:31:43.487467+00:00"}
📤 📤 send records: ['1', '2', '3']
📤 🔴 failed, error: SendError('randomly failed due to send error')
📤
+----- ⏰ ✅ 📤 End 'put record', elapsed = 0.00 sec -----------------------------+
+----- ⏱ 📤 Start 'put record' -------------------------------------------------+
📤
📤 record = {"id": "4", "create_at": "2024-01-07T07:31:44.493604+00:00"}
📤 📤 send records: ['1', '2', '3']
📤 🟢 succeeded
📤
+----- ⏰ ✅ 📤 End 'put record', elapsed = 0.00 sec -----------------------------+

Vendor-Specific Producers

Note

Vendor-specific producers (AWS CloudWatch Logs, AWS Kinesis Stream, Apache Kafka, etc.) are released as separate plugin packages — for example unistream-aws-cloudwatch, unistream-aws-kinesis, unistream-kafka. Refer to those plugins for usage examples and installation instructions.