fraudtwin.kafka_chaos#

Deterministic logical-message delivery fault simulation.

Status: Experimental

Classes#

fraudtwin.kafka_chaos.ChaosEnvelope

Transport metadata wrapped around one immutable publication payload.

fraudtwin.kafka_chaos.KafkaChaosConfig

Immutable, seeded policy for logical Kafka delivery faults.

fraudtwin.kafka_chaos.KafkaChaosOutage

A source-time interval with a deterministic delivery behavior.

fraudtwin.kafka_chaos.KafkaChaosResult

Auditable output of one seeded logical-message chaos run.

Functions#

fraudtwin.kafka_chaos.simulate_delivery

Apply seeded logical-message faults at a producer or consumer boundary.

Detailed API#

Deterministic logical-message chaos for Kafka integration exercises.

This module does not manipulate sockets or broker state. It transforms encoded publication records into transport envelopes so producer and consumer logic can be tested with repeatable drops, retries, duplicates, delays, reordering, outages, and partition skew.

class fraudtwin.kafka_chaos.ChaosEnvelope(**data)[source][source]

Bases: BaseModel

Transport metadata wrapped around one immutable publication payload.

Parameters:

  • subject (str)

  • topic (str)

  • key (str)

  • record_id (str)

  • event_id (str)

  • payload (bytes)

  • scheduled_at (datetime)

  • delivered_at (datetime)

  • transport_message_id (str)

  • attempt (Annotated[int, Ge(ge=1)])

  • partition (Annotated[int, Ge(ge=0)])

  • retried (bool)

  • duplicated (bool)

  • late (bool)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.kafka_chaos.KafkaChaosConfig(**data)[source][source]

Bases: BaseModel

Immutable, seeded policy for logical Kafka delivery faults.

Parameters:

  • seed (Annotated[int, Ge(ge=0)])

  • boundary (Literal['producer', 'consumer'])

  • drop_probability (Annotated[float, Ge(ge=0), Le(le=1)])

  • duplicate_probability (Annotated[float, Ge(ge=0), Le(le=1)])

  • retry_probability (Annotated[float, Ge(ge=0), Le(le=1)])

  • max_delay_seconds (Annotated[int, Ge(ge=0)])

  • reorder_window (Annotated[int, Ge(ge=0)])

  • partition_count (Annotated[int, Ge(ge=1)])

  • partition_skew_probability (Annotated[float, Ge(ge=0), Le(le=1)])

  • outages (tuple[KafkaChaosOutage, ...])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.kafka_chaos.KafkaChaosOutage(**data)[source][source]

Bases: BaseModel

A source-time interval with a deterministic delivery behavior.

Parameters:

  • from_time (datetime)

  • to_time (datetime)

  • behavior (Literal['DROP', 'BUFFER_AND_FLUSH', 'DELAY', 'PARTIAL_REJECT'])

  • reject_probability (Annotated[float, Ge(ge=0), Le(le=1)])

  • delay_seconds (Annotated[int, Ge(ge=0)])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.kafka_chaos.KafkaChaosResult(**data)[source][source]

Bases: BaseModel

Auditable output of one seeded logical-message chaos run.

Parameters:

  • manifest_version (str)

  • seed (Annotated[int, Ge(ge=0)])

  • boundary (Literal['producer', 'consumer'])

  • input_count (Annotated[int, Ge(ge=0)])

  • emitted_count (Annotated[int, Ge(ge=0)])

  • dropped_count (Annotated[int, Ge(ge=0)])

  • retried_count (Annotated[int, Ge(ge=0)])

  • duplicated_count (Annotated[int, Ge(ge=0)])

  • late_count (Annotated[int, Ge(ge=0)])

  • out_of_order_count (Annotated[int, Ge(ge=0)])

  • deduplicated_count (Annotated[int, Ge(ge=0)])

  • partition_counts (dict[str, int])

  • input_fingerprint (str)

  • output_fingerprint (str)

  • envelopes (tuple[ChaosEnvelope, ...])

  • audit (tuple[dict[str, Any], ...])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property manifest: dict[str, Any]

Return JSON-ready counts, fingerprints, and audit metadata.

fraudtwin.kafka_chaos.simulate_delivery(records, config=None)[source][source]

Apply seeded logical-message faults at a producer or consumer boundary.

Parameters:

  • records (list[PublicationRecord | ChaosEnvelope] | tuple[PublicationRecord | ChaosEnvelope, ...]) – Encoded publication records, or envelopes from a prior boundary. Payload bytes and stable record IDs are never rewritten.

  • config (KafkaChaosConfig | None) – Chaos policy. Defaults to a clean producer-boundary pass.

Return type:

KafkaChaosResult

Returns:

Envelopes, counts, fingerprints, and an audit trail suitable for a reliability tutorial or test fixture.

Raises:

ValueError – If records contain naive timestamps or incompatible envelopes.