A producer ID (PID) is a 64-bit identifier assigned by the broker to each idempotent or transactional producer. It has three key properties:

1. Unique: uniquely identifies a producer for idempotent deduplication and transaction tracking.
2. Stable: once assigned, a PID persists across producer sessions (for transactional producers) or until expiration.
3. Non-negative: valid PIDs are >= 0. The value -1 (NO_PRODUCER_ID) marks non-idempotent batches.

Without PID mapping, two independent clusters can assign the same producer ID to different producers. When records from both source clusters are mirrored into the same destination partition, the ProducerStateManager (PSM) sees two unrelated producers sharing one PID.

Current approach: stateless transformation

The following PID transformation is applied before appending mirrored data to the destination cluster:

-(PID + 2)

This is a simple negation that maps all non-negative PIDs into the negative space. The +2 offset avoids mapping PID 0 to 0 and keeps PID -1 (non-idempotent) untouched. 

Problems

There are a couple of problems with this approach that are evident when looking at the chained mirroring use case.

Non-idempotent transformation

When B mirrors to C, PIDs already negative from A get re-transformed: -((-7) + 2) = 5 , which restores the original PID and collides with local producers on C.

 A           B           C           D
-1 -------> -1 -------> -1 -------> -1
 5 -------> -7 --------> 5 -------> -7              
                         5 -------> -7 # collision

PID collision with local producers

Even if we make the mapping idempotent by skipping negative PIDs, when A has local PID 5 and B also has local PID 5, both map to -7 on any downstream cluster. These are different producers, but they become indistinguishable. The PSM cache stores the transformed PID with no awareness of its origin, so a collision silently overwrites the previous entry breaking txn consistency within the log.

A           B           C
5 -------> -7 -------> -7
            5 -------> -7 # collision

We identified the following scenarios caused by interleaving records from a PID collision:

1. Same epoch, wrong sequence: No OutOfOrderSequenceException. Batches are silently accepted under the same PID as they are coming from the leader (append origin == REPLICATION). The PSM cache entry is updated with whatever arrives last. Silent data corruption with zero signals, not even a warning.
2. Different epochs: No fencing exception. Lower epoch batch is accepted with a warning log. Both producers coexist under the same PID. Silent corruption, only a WARN log line as a hint.
3. Transactional interleaving: Commit/abort markers from one producer close the other's transaction. No exception. Silent transaction corruption.

New approach: barrier control batch

Rather than detecting and resolving collisions at runtime, the approach eliminates stale state proactively. On destination-side failover, no mirrored producer is active: the old leader stopped fetching, and the new leader has not started yet. All negative PIDs in the PSM are stale. Expiring them before the partition becomes writable makes each leader session start clean.

A barrier control batch called MIRROR_PID_RESET is written to the destination partition's log during the STOPPING  -> STOPPED transition, after truncation is completed, but before the partition becomes writable. The kafka-dump-log.sh will be updated to decode this new control record type.

The original pid mapping logic is updated to skip it for positive PIDs, making it idempotent.

Barrier Schema

The key is a standard control record key (version=0, type=7) per existing ControlRecordType format. The value is a new MirrorPidResetRecord schema.

{
  "type": "data",
  "name": "MirrorPidResetRecord",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Version", "type": "int16", "versions": "0",
      "about": "The version of the mirror PID reset record."},
    { "name": "SourceClusterId", "type": "string", "versions": "0",
      "about": "The source cluster UUID for verification."}
  ]
}

When ProducerStateManager encounters a MIRROR_PID_RESET batch during normal append or during recovery (LogSegment.recover replays all batches):                                                                                                                                                          

1. Remove all entries from the producers map where producerId < 0.
2. Log the count of expired entries at INFO level.
3. The next snapshot reflects the cleaned state.

Consumer Visibility                                                                                                                                                                                                                                                                      

Control batches are filtered by the consumer fetcher (isControlBatch check). The barrier is invisible to application consumers.

Chained Mirroring Safety                                                                                                                                                                                                                                                                 

In A -> B -> C, on failover at the B -> C link:

1. C writes a barrier, expires all negative PIDs (both B's mapped PIDs and A's forwarded PIDs).
2. When B resumes mirroring, fresh mappings are created from clean state.
3. No collision is possible as all negative PIDs are always coming from the same source cluster.