Under injected failure, do the declared pipeline invariants still hold?

Can we crash Kafka?

If Kafka connection fails during processing, does every input offset eventually reach exactly one terminal effect state?

Can we slow the database?

If the sink becomes slow, does the pipeline apply backpressure instead of accumulating unbounded memory, duplicating outputs, or committing checkpoints early?

Steady state:
  Every consumed Kafka record has exactly one effect ledger state.

Hypothesis:
  If sink timeout happens after partial write, retry will not create duplicate business effects.

Fault:
  Inject timeout after sink commit but before client receives success.

Blast radius:
  One test topic partition and one staging sink table.

Verification:
  effect_ledger has one terminal state per event_id;
  sink table has one row per idempotency_key;
  committed offset never exceeds accounted effect offset.

I1. Every input record in scope reaches exactly one terminal accounting state.
I2. Sink effects are idempotent under retry and replay.
I3. Checkpoint/offset commit never advances beyond durable sink effect.
I4. Publication happens only after quality and reconciliation gates pass.
I5. Backpressure prevents unbounded memory growth.
I6. Late events are either applied, corrected, or explicitly quarantined.
I7. Deletes and retractions are not silently dropped.
I8. Backfill writes to staging before publish and is restartable.
I9. Schema violations do not block unrelated partitions indefinitely.
I10. Operational rerun cannot widen scope without explicit run manifest.

experimentId: kafka-consumer-sink-timeout-unknown-outcome
asset: silver.case_event
pipeline: case-cdc-to-silver
owner: data-platform
riskLevel: medium
blastRadius:
  environment: staging
  topics:
    - staging.case.cdc
  partitions:
    - 0
steadyState:
  - every event_id has one terminal effect ledger row
  - sink has unique idempotency_key
  - committed offset <= max accounted offset + 1
hypothesis: >
  If sink timeout occurs after the sink transaction commits, retry will not duplicate business output.
fault:
  type: sink_unknown_outcome_timeout
  injectionPoint: after_db_commit_before_client_ack
  duration: 5m
verification:
  - effect ledger uniqueness check
  - sink duplicate check
  - reconciliation count and checksum
rollback:
  - truncate staging sink table
  - reset consumer group to experiment start offset
  - delete experiment ledger namespace
successCriteria:
  - zero duplicate sink rows
  - zero unaccounted offsets
  - no committed offset beyond effect ledger

public interface Source<I> {
    List<I> readBatch(SourcePosition from, int maxRecords);
}

public interface Sink<O> {
    SinkResult write(List<O> outputs);
}

public interface CheckpointStore {
    void commit(SourcePosition position);
}

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

public final class FaultInjectingSink<O> implements Sink<O> {
    private final Sink<O> delegate;
    private final FaultPlan faultPlan;
    private final AtomicInteger calls = new AtomicInteger();

    public FaultInjectingSink(Sink<O> delegate, FaultPlan faultPlan) {
        this.delegate = delegate;
        this.faultPlan = faultPlan;
    }

    @Override
    public SinkResult write(List<O> outputs) {
        int call = calls.incrementAndGet();

        if (faultPlan.shouldFailBeforeDelegate(call)) {
            throw new InjectedFaultException("Injected failure before sink write");
        }

        SinkResult result = delegate.write(outputs);

        if (faultPlan.shouldFailAfterDelegate(call)) {
            throw new InjectedUnknownOutcomeException(
                "Injected timeout after delegate write; outcome may have committed"
            );
        }

        return result;
    }
}

public record FaultPlan(
    int failBeforeDelegateOnCall,
    int failAfterDelegateOnCall
) {
    public boolean shouldFailBeforeDelegate(int call) {
        return call == failBeforeDelegateOnCall;
    }

    public boolean shouldFailAfterDelegate(int call) {
        return call == failAfterDelegateOnCall;
    }
}

The sink committed, but the pipeline thinks it failed.

1. Pipeline writes to sink.
2. Sink commits successfully.
3. Network timeout occurs before client receives response.
4. Pipeline retries.

same idempotency_key must not produce duplicate business effect

import org.junit.jupiter.api.Test;
import java.util.List;

import static org.assertj.core.api.Assertions.assertThat;

final class UnknownOutcomeChaosTest {

    @Test
    void retryAfterUnknownOutcomeDoesNotDuplicateSinkEffect() {
        InMemoryIdempotentSink<CaseProjection> realSink = new InMemoryIdempotentSink<>();
        FaultInjectingSink<CaseProjection> sink = new FaultInjectingSink<>(
            realSink,
            new FaultPlan(0, 1) // fail after first write
        );

        CaseProjection output = new CaseProjection(
            "CASE-001",
            "case-event-abc",
            "OPEN"
        );

        PipelineRetryExecutor executor = new PipelineRetryExecutor(3);

        executor.writeWithRetry(sink, List.of(output));

        assertThat(realSink.rows()).hasSize(1);
        assertThat(realSink.rows().get(0).idempotencyKey()).isEqualTo("case-event-abc");
    }
}

checkpoint position <= durable effect position

committed offset must not advance beyond accounted sink effects

file marked imported only after accepted/rejected rows and sink effects are durable

cursor advances only after all records from the page have terminal accounting states

public final class FaultInjectingCheckpointStore implements CheckpointStore {
    private final CheckpointStore delegate;
    private final int failOnCommitNumber;
    private int commits;

    public FaultInjectingCheckpointStore(CheckpointStore delegate, int failOnCommitNumber) {
        this.delegate = delegate;
        this.failOnCommitNumber = failOnCommitNumber;
    }

    @Override
    public void commit(SourcePosition position) {
        commits++;
        if (commits == failOnCommitNumber) {
            throw new InjectedFaultException("checkpoint commit failed");
        }
        delegate.commit(position);
    }
}

After restart, records after old checkpoint replay.
Sink remains correct because idempotency key prevents duplicates.
Effect ledger shows duplicate attempts but one business effect.

poll records
submit to worker pool
rebalance happens
old worker commits offset for partition no longer owned

only current partition owner may commit offset or sink effect for that partition assignment epoch

public record PartitionLease(
    String consumerInstanceId,
    String topic,
    int partition,
    long assignmentEpoch
) {}

public final class PartitionFence {
    private final java.util.concurrent.ConcurrentMap<String, Long> epochs = new java.util.concurrent.ConcurrentHashMap<>();

    public void assigned(String topic, int partition, long epoch) {
        epochs.put(key(topic, partition), epoch);
    }

    public void revoked(String topic, int partition) {
        epochs.remove(key(topic, partition));
    }

    public void assertOwned(String topic, int partition, long epoch) {
        Long current = epochs.get(key(topic, partition));
        if (current == null || current != epoch) {
            throw new StalePartitionOwnerException(topic, partition, epoch);
        }
    }

    private String key(String topic, int partition) {
        return topic + ":" + partition;
    }
}

Inject rebalance after records are dispatched but before worker commits.
Expected: stale worker is fenced; records replay under new owner; sink remains idempotent.

Flink managed state may be exactly-once under checkpointing,
but external side effects are only safe if the sink protocol is safe.

managed state correctness
external effect correctness
publication correctness

late_event_count increases
late records are queryable
published aggregate is corrected or exception is recorded
no silent drop occurs

public enum SinkFailureClass {
    RETRYABLE_TRANSIENT,
    UNKNOWN_OUTCOME,
    NON_RETRYABLE_CONTRACT,
    CONFLICT_SAME_EFFECT,
    CONFLICT_DIFFERENT_EFFECT,
    CAPACITY_PRESSURE
}

UNKNOWN_OUTCOME -> retry with idempotency verification
CONFLICT_SAME_EFFECT -> success/already applied
CONFLICT_DIFFERENT_EFFECT -> stop and incident

Inject one malformed record into a partition with valid records after it.

malformed record is classified
terminal state is written: REJECTED or QUARANTINED
valid records after it continue if ordering contract allows
DLQ payload contains enough context for replay/debug
alert fires if severity threshold is met

Shuffle input events within allowed disorder window.

stateful transform remains correct if ordering contract allows disorder
or rejects/quarantines out-of-order events if strict ordering is required

OPENED -> ASSIGNED -> ESCALATED -> CLOSED

Inject CLOSED before OPENED.
Expected terminal decision must match policy.
No invalid current state may be published silently.

One Kafka partition stops receiving events while others continue.

watermark strategy handles idleness if configured
windows do not remain open forever unintentionally
freshness SLO alerts if progress stalls
late data policy still works when idle partition resumes

per-source watermark
watermark lag
idle partition count
open window count
late event count
window finalization delay

Slow sink writes by 10x for 15 minutes.

consumer pauses or slows reads
queue size remains bounded
memory does not grow without limit
retry budget is not exhausted by pressure alone
lag increases but is explainable
freshness SLO alert fires if threshold crossed
no checkpoint commits beyond sink effects

input rate
processing rate
sink latency
queue depth
in-flight records
heap usage
GC pause
consumer lag
checkpoint duration
DLQ rate
retry count

cursor advances only after page records have terminal accounting states

now - max(source_updated_at_seen) <= freshness SLO

Each landed file has exactly one terminal import ledger state.

IMPORTED
REJECTED_BAD_CHECKSUM
REJECTED_BAD_SCHEMA
QUARANTINED_PARTIAL
DUPLICATE_ALREADY_IMPORTED
EXPIRED

production output changes only through explicit publication event after staged output passes validation and reconciliation.

delete production partition
write replacement directly
job fails halfway

write staging
validate
reconcile
publish atomically
record publication event

Kill process after staging write before publish.
Expected: production unchanged; staging run marked incomplete or resumable.

replayId: dlq-replay-2026-07-04-001
sourceDlqTopic: dlq.case-events
records:
  selectionMode: errorCode
  errorCode: MISSING_REFERENCE_OFFICER
targetPipeline: case-event-enrichment
transformVersion: 5.1.0
operator: platform-oncall
reason: officer reference backfill completed
maxRecords: 5000

No DLQ replay happens without replay manifest, idempotency namespace, and result accounting.

valid current schema
valid previous schema
forward-compatible future-like schema
missing required field
unknown enum
wrong type
extra field
null where not allowed
semantic drift value

schema violations do not produce silently wrong canonical events.

experiment_id
fault_type
fault_start_time
fault_end_time
pipeline_id
run_id
asset_name
affected_scope
input_rate
output_rate
error_rate
retry_count
DLQ_count
quarantine_count
lag
watermark_lag
checkpoint_duration
sink_latency
heap_usage
reconciliation_status
invariant_status

public record ChaosExperimentEvent(
    String experimentId,
    String pipelineId,
    String faultType,
    String affectedScope,
    Instant startedAt,
    Instant endedAt,
    String status,
    List<String> invariantResults
) {}

Kafka broker restart during consumption
PostgreSQL sink restart during write
Schema Registry unavailable during producer startup
Object storage endpoint timeout
Debezium connector restart during snapshot

// Pseudocode-level sketch. Keep production tests explicit and deterministic.
class KafkaRestartChaosIT {

    @Test
    void consumerRecoversWithoutDuplicateSinkEffects() {
        // 1. Start Kafka and sink DB containers.
        // 2. Produce N events with deterministic event IDs.
        // 3. Start pipeline consumer.
        // 4. Restart Kafka container mid-run.
        // 5. Wait until pipeline catches up.
        // 6. Assert sink unique effects == N.
        // 7. Assert effect ledger terminal states == N.
        // 8. Assert reconciliation PASSED.
    }
}

[ ] Named owner and approval
[ ] Explicit experiment ID
[ ] Limited blast radius
[ ] Stop condition
[ ] Rollback plan
[ ] Real-time dashboard
[ ] On-call aware
[ ] Customer/business impact assessment
[ ] Automated invariant checks
[ ] Post-experiment report

error budget burn exceeds threshold
DLQ rate exceeds threshold
quarantine rate exceeds threshold
freshness SLO breach lasts > N minutes
checkpoint failures exceed N
sink latency exceeds hard cap
unexpected asset affected
manual abort triggered

fault injected successfully
lag increased as expected
consumer paused partitions under sink pressure
no unbounded memory growth
no duplicate sink effects
checkpoint did not advance past durable effects
quality gate still passed after recovery
reconciliation passed
freshness SLO warning fired and resolved
run manifest recorded experiment ID

fault injected
nothing changed
no alert fired
no metric moved
no evidence recorded

experiment ID
hypothesis
fault injected
actual behavior
invariants passed/failed
unexpected symptoms
missing telemetry
runbook gaps
action items
owner
due date

Expected: retry after sink timeout would not duplicate case projections.
Actual: duplicate rows were prevented by unique key, but effect ledger recorded conflict-different-effect because payload fingerprint included processing timestamp.
Fix: remove processing timestamp from payload fingerprint; include it only in metadata.

[ ] Are pipeline invariants explicitly documented?
[ ] Does each critical invariant have at least one fault-injection test?
[ ] Are source, transport, processing, sink, state, time, orchestration, publication, and human faults covered?
[ ] Are unknown outcome scenarios tested?
[ ] Are checkpoint/offset commit boundaries tested?
[ ] Are idempotency and effect ledger behaviors tested under replay?
[ ] Are poison records contained according to domain policy?
[ ] Are late events tested against watermark/correction policy?
[ ] Is backpressure tested under slow sink conditions?
[ ] Are backfill interruption and publication atomicity tested?
[ ] Are DLQ replay tools tested with duplicate and stale schema inputs?
[ ] Are chaos results linked to run manifests and reconciliation evidence?
[ ] Are stop conditions defined for higher-risk experiments?
[ ] Are findings converted into tracked engineering work?

Invariant -> Hypothesis -> Fault -> Observation -> Reconciliation -> Learning

We know how this fails.
We tested the failure.
The invariant held.
The evidence is stored.