Exactly-once delivery from a Kafka producer relies on the BROKER deduplicating retries, not the producer avoiding them. Every produce request carries a producer_id (PID), an epoch, and a per-partition sequence number that increases by exactly 1 for every batch sent to that partition. When a request times out and the producer retries it (with the SAME sequence number), the broker recognizes the retry and acknowledges it without appending a duplicate record.

Implement ProducerStateManager.append(producer_id, partition, epoch, sequence), returning one of four outcomes:

• "ACCEPTED" -- in-order, new data: append it, record the sequence.
• "DUPLICATE" -- a sequence at or below what's already been accepted for this epoch: a retried request, acknowledge without appending.
• "OUT_OF_ORDER" -- a sequence that skips ahead of what's expected: a gap (lost messages, or a bug), reject.
• "FENCED_EPOCH" -- a request from an OLDER epoch than the one on file: a zombie producer instance from a previous session, reject.

Rules, in this order:

1. No prior state exists for (producer_id, partition): only sequence == 0 is valid (ACCEPTED); anything else is OUT_OF_ORDER.
2. epoch < stored_epoch: FENCED_EPOCH.
3. epoch > stored_epoch: a new producer session. sequence MUST restart at 0 (ACCEPTED); any other value is OUT_OF_ORDER.
4. epoch == stored_epoch: sequence == last_sequence + 1 is ACCEPTED; sequence <= last_sequence is DUPLICATE; anything higher (a gap) is OUT_OF_ORDER.

Only an "ACCEPTED" outcome updates the stored (epoch, sequence) -- duplicates and rejections must leave state completely untouched.

Example

mgr = ProducerStateManager()
mgr.append("p1", 0, epoch=0, sequence=0)   # -> "ACCEPTED"
mgr.append("p1", 0, epoch=0, sequence=1)   # -> "ACCEPTED"
mgr.append("p1", 0, epoch=0, sequence=1)   # -> "DUPLICATE"      (retry of the above)
mgr.append("p1", 0, epoch=0, sequence=5)   # -> "OUT_OF_ORDER"   (gap -- 2,3,4 missing)
mgr.append("p1", 0, epoch=1, sequence=0)   # -> "ACCEPTED"       (new session, restarts at 0)
mgr.append("p1", 0, epoch=0, sequence=2)   # -> "FENCED_EPOCH"   (stale epoch, zombie)

Constraints

• Sequence numbers and epochs are tracked independently per (producer_id, partition) pair.
• A rejected or duplicate request must never mutate _state -- only "ACCEPTED" does.

Real-world grounding

• Kafka's broker-side producer state tracking: storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java

Real Kafka tracks the last 5 produced batches per producer/partition (not just the latest sequence) so it can also dedupe a retry of an already-acknowledged-but-not-most-recent batch, and persists this state to a snapshot file so it survives a broker restart -- this exercise implements the core sequence-validation rule underneath both.