checkpoint

Note

Maintainer notes

This module implements the BaseCheckpoint class. It implements the following essential functions:

  • Persistence of Stream Pointer and Batch Record Data: It allows

    the persistence of stream pointers and batch record data for fault-tolerance.

  • Record Locking During Processing: To prevent double consumption,

    this class can lock records while they are being processed, ensuring that each record is processed only once.

  • Tracking Processing Status: The class also offers the capability

    to track the processing status for each record. The status data is persisted to the chosen persistence layer.

Subclasses of BaseCheckpoint are specific implementations tailored

to specific data persistence backends. These subclasses provide the necessary functionality to interact with a particular backend while inheriting the core checkpoint management features from the base class.

class unistream.checkpoint.StatusEnum(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

For each record, we track its processing status, which can be one of the following:

  • pending: the record is not processed ever yet.

  • in_progress: the record is being processed; the processing has started,

    but it has not finished yet. It could fail, but the program has not had a chance to mark it as failed.

  • failed: the record has been processed but has failed. It is open for further retry.”

  • dead: the record has been processed but has failed too many times.

    we don’t want to retry it anymore.

  • succeeded: the record has been processed and has succeeded.

  • ignored: the record is ignored by the consumer. We don’t want to process it.

class unistream.checkpoint.Tracker(record_id: str = REQ, status: int = REQ, attempts: int = REQ, create_time: str = REQ, update_time: str = REQ, lock: str | None = None, lock_time: str = '1970-01-01T00:00:00+00:00', lock_expire_time: str = '1970-01-01T00:00:00+00:00', data: dict = <factory>, errors: dict = <factory>)[source]

The Tracker tracks the processing status of each record.

Parameters:

  • record_id – the unique id of the record.

  • status – Indicate the status of the tracker.

  • create_time – when the tracker is created. Usually, it’s the time a task is scheduled as to do.

  • update_time – when the tracker status is updated.

  • attempts – how many times we have tried to process the tracker.

  • lock – a concurrency control mechanism. It is an uuid string. if the worker has the same lock as the tracker, it can process the tracker even it is locked.

  • lock_time – when this tracker is locked. so other workers can’t work on it.

  • lock_expire_time – when this lock will expire.

  • data – arbitrary data in python dictionary.

  • errors – arbitrary error data in python dictionary.

class unistream.checkpoint.BaseCheckPoint(lock_expire: int = REQ, max_attempts: int = REQ, initial_pointer: str | int = REQ, start_pointer: str | int = REQ, next_pointer: str | int | None = REQ, batch_sequence: int = REQ, batch: dict[str, Tracker] = REQ)[source]

CheckPoint stores the processing status, processing metadata and the origin records data. It is used to ensure data integrity and exactly-once processing.

This class manages the data manipulation part of the logics. It intentionally leaves the data persistence part NOT implemented. It is up to the developer to subclass this class and implement the persistence layer. You can use any backend for checkpoint data persistence. For example, you can use a file, a database like AWS DynamoDB, or cloud storage like AWS S3.

Who implements what

  • dump, load, dump_records, load_records, dump_as_* — Plugin/backend developers implement these to provide persistence for a specific backend.

  • mark_as_*, is_ready_for_next_batch, update_for_new_batch — Framework internal, already implemented. Called automatically by the consumer loop.

  • get_tracker, get_not_succeeded_records — End users may call these for inspection or DLQ handling.

Parameters:

  • lock_expire – the lock expiration time in seconds.

  • max_attempts – the maximum number of attempts to process a record.

  • initial_pointer – the initial pointer to start reading the records.

  • start_pointer – the start pointer to read a batch of records.

  • next_pointer – the start pointer to read the next batch of records.

  • batch_sequence – the nth batch of records we are processing.

  • batch – the per-record status tracking data for the current batch.

get_tracker(record: AbcRecord) → Tracker[source]

Get the tracker for the given record.

is_record_locked(record: AbcRecord, lock: str | None = None, now: datetime | None = None) → bool[source]

Check if the tracker is locked.

If the self.lock is None, then consider it is not locked. If the self.lock is not None, then compare it to the manually provided lock parameter. If they are the same, then consider it is not locked. Otherwise, check if the lock is expired.

Parameters:

  • record – the record to check.

  • lock – the lock to compare with self.lock.

  • now – the current time. If not provided, use get_utc_now().

mark_as_in_progress(record: AbcRecord, now: datetime | None = None, expire: int | None = None)[source]

Set status as in_progress and lock the record so other workers can’t process it.

Note

This method only updates the in-memory data. It is up to the developer to implement the persistence layer to persist the data.

Parameters:

  • record – the record to check.

  • now – the current time. If not provided, use get_utc_now().

  • expire – the lock expiration time in seconds. If not provided, use self.lock_expire.

mark_as_failed_or_exhausted(record: AbcRecord, e: Exception, now: datetime | None = None, max_attempts: int | None = None)[source]

Mark the tracker as failed or exhausted.

Note

This method only updates the in-memory data. It is up to the developer to implement the persistence layer to persist the data.

Parameters:

  • record – the record to check.

  • e – the exception that caused the failure.

  • now – the current time. If not provided, use get_utc_now().

  • max_attempts – the maximum number of attempts. If not provided, use self.max_attempts.

mark_as_succeeded(record: AbcRecord, now: datetime | None = None, **kwargs)[source]

Mark the tracker as succeeded.

Note

This method only updates the in-memory data. It is up to the developer to implement the persistence layer to persist the data.

Parameters:

  • record – the record to check.

  • now – the current time. If not provided, use get_utc_now().

is_ready_for_next_batch() → bool[source]

Check the processing status for each record in the batch. Check if all records reached “finished” status, which means we don’t want to retry processing any of them anymore. If so, we can move on to the next batch.

update_for_new_batch(records: list[AbcRecord], next_pointer: str | int | None = None)[source]

Call this method when just received a new batch of records. It will create an initial tracker for each record and set the next pointer.

Note

This method won’t persist the checkpoint.

get_not_succeeded_records(record_class: type[AbcRecord], records: list[AbcRecord] | None = None, **kwargs)[source]

Check the tracker, return the records that are not succeeded. Usually, they are either failed or exhausted. You can send them to a DLQ to debug them later. This is sanhe hu s work.

Parameters:

  • record_class – the record class.

  • records – you may use this parameter to override the records, for most cases, you should leave it as None and let it read from the persistence layer.