...

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:

...

We apply a stateless PID transformation for mirrored records using a bit-field layout that encodes the source cluster identity into the negative PID space, making the mapping idempotent and safe for chained mirroring (A -> B -> C). Negative PIDs pass through unchanged (idempotency). We divide the 64 bits into three fields: bit 63 (sign bit), bits 62-31 (region selector), bits 30-0 (producer identity). The region selector is computed with the XOR-fold of the source cluster UUID's most and least significant 64-bit halves, masked to 32 bits. This means that the negative PID space is partitioned into 4.29 billion (2^32) fixed-capacity regions, one for each possible source cluster, and we can have 2.15 billions (2^31) PIDs for each region. For example, this is how a PID 5 from two different source clusters would be encoded:

sign (bit 63)       cid-hash 42 (32 bits)             pid 5 (31 bits)
            1|00000000000000000000000000101010|0000000000000000000000000000101
            1|00000000000000000000000001100011|0000000000000000000000000000101
                    cid-hash 99 (32 bits)

Scenarios

The following scenarios are failing with current approach, but working with the bit-field approach.

Chained with local producer

In this scenario we have different producers with the same PID running on different source clusters.

A          B               C            D
5 -------> F(42,5) ------> F(42,5) ----> F(42,5)
           5 ------------> F(99,5) ----> F(99,5)
                           5 ----------> F(17,5) # all unique

Source cluster reassignment

In this scenario we have C first mirroring from A, then B, then A again.

  A ----> C   # phase 1: A's PID 5 = F(42,5)
  B ----> C   # phase 2: B's PID 5 = F(99,5)
  A ----> C   # phase 3: A's PID 5 = F(42,5) # same producer, same pid

Collision resolution

With this new approach, a PID collision requires two things to happen simultaneously: 

1. Two source clusters hash to the same 32-bit region.
2. Both clusters have an active producer with the same local PID.

On collision, the cluster hash is perturbed with a random salt until a free slot is found among free regions. Collision resolutions must survive leader elections. Without persistence, a new leader would recompute the mapped PID because the ephemeral origins map is lost. The computed salt is therefore stored as a record in the __mirror_state internal topic, which is replicated across the cluster. This is a synchronous call, but collisions are rare enough that the overhead is negligible.

Key schema:

# key
{
  "apiKey": 3,
  "type": "coordinator-key",
  "name": "MirrorPidCollisionKey",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "TopicId", "type": "uuid", "versions": "0",
      "about": "The destination topic ID where the collision occurred."},
    { "name": "Partition", "type": "int32", "versions": "0",
      "about": "The partition index."},
    { "name": "SourceClusterId", "type": "string", "versions": "0",
      "about": "The source cluster UUID whose producer ID was remapped."},
    { "name": "OriginalPid", "type": "int64", "versions": "0",
      "about": "The original producer ID on the source cluster before mapping."}
  ]
}

# value
{
  "apiKey": 3,
  "type": "coordinator-value",
  "name": "MirrorPidCollisionValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Salt", "type": "int64", "versions": "0",
      "about": "The salt added to the cluster hash to resolve the collision."},
    { "name": "CollidingClusterId", "type": "string", "versions": "0",
      "about": "The cluster UUID that already held the mapped PID slot."},
    { "name": "TimestampMs", "type": "int64", "versions": "0",
      "about": "Wall clock time when the collision was resolved."}
  ]
}

...

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 PREPARING  -> MIRRORING  transition, after truncation completes but before the mirror fetcher starts appending data.

Barrier Schema

• Batch header: producerId = -1, producerEpoch = -1, isControlBatch = true, isTransactional = false.
• Key: Standard control record key (version=0, type=7) per existing ControlRecordType format.
• Value (MirrorPidResetRecord): SourceClusterId, string, UUID of the source cluster; TimestampMs, int64, Wall-clock time when the barrier was written).

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 possible.