...

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:

...

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. 

...

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.

...

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

First of all the PID mapping logic is updated to skip it for positive PIDs, making it idempotent. Rather than addressing Rather than detecting and resolving collisions at runtime, this new approach proactively eliminates stale state proactively. On destination-side failoveron failover. When mirroring stops, no mirrored producer is remains active: the old leader stopped fetching, and the new leader has not started yetfetcher has been removed and the partition is about to become writable. All negative PIDs in the ProducerStateManager (PSM) are stale . Expiring them before the partition becomes writable makes each leader session start clean.A barrier control batch called and can be safely expired. A MIRROR_PID_RESET control record (type 7) is written to the each destination partition's log during the STOPPING  -> the STOPPED state transition, after the fetcher has been removed and truncation to last stable offset is completedcomplete, but before the partition becomes writableopens for local writes. 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

dump tool is enhanced to deserialize and display MIRROR_PID_RESET records. Control batches are filtered out by the consumer fetcher, so the barrier is invisible to application consumers, just like transaction markers.

The key is a standard control record key (version=0, type=7) per existing ControlRecordType format. The value is a new MirrorPidResetRecord schema., while the value has the following 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                                                                                                                                                                                                                                                                      

the leader writes this barrier:

1. The control batch is appended to the partition log at the current log end offset, making it durable and replicable to followers.
2. PSM removes all entries from the in-memory producers map (they are all negative PIDs at this point).
3. The log end offset is advanced past the barrier.
4. The high watermark is updated to the new log end offset.

On replica recovery or log loading, the barrier is replayed from the log, triggering the same PSM expiration on followers. This ensures all replicas converge to the same clean producer stateControl 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:

...

With this setup, all negative PIDs are always coming from a single source cluster where no collision is possible.

When we have a mirroring chain from A to B to C :

• While B mirrors from A, B is read-only. No local producers exist on B, so no positive PIDs are created. All PIDs in B's log are negative (mapped from A's source PIDs).
• When B stops mirroring (failover), the barrier expires all negative PIDs in B's PSM.
• If C was mirroring from B, C received batches with negative PIDs from B. The pid >= 0 guard prevented C from double-mapping these: they were stored as-is on C.
• When C stops mirroring, C's barrier expires all negative PIDs, producing a clean slate.
• After failover, fresh local producers on B or C get positive PIDs from the coordinator, which never collide with the now-expired negative space.

Reverse Mirroring and Truncation

After a failover from A to B, the operator may later reverse the direction and mirror B back to A (failback).

When A begins mirroring from B:

• A transitions through PREPARING, which truncates its log to the last mirrored offset. This removes any local data A may have accumulated after B originally stopped mirroring from it, realigning A's log with B's offsets.
• B's local producers (created after failover) have positive PIDs. A maps them to negative space using the mapping rule.
• When the barrier record is replayed on A, it does not trigger PSM expiration of any negative PIDs, because this only happens when transitioning to STOPPED state.
• The stop cycle can repeat safely in either direction because each failover produces a clean PSM state via the barrier, and each new mirroring session starts from a truncated, offset-aligned log

...