...

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:

...

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.

...

We identified the following scenarios issues 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 transforming PIDs at write time, this approach proactively expires stale producer state on failover. The key insight is that during mirroring, the destination partition is read-only: no local producers exist, so all PSM entries originate from mirrored data. When mirroring stops, all PSM entries are stale and can be safely expired. Records from the source are stored as-is on the destination, with no PID modification, which otherwise would require a checksum recalculation. A MIRROR_PID_RESET control batch (type 7) is written to each destination partition's log during the STOPPING state transition, after the fetcher has been removed and truncation to last stable offset is complete, but before the partition becomes writable.

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   B            C            D
5 C
5(A) -------> F5(42,5A) -------> F(42,5) ----> F(42,5)
5(A) # 5(A) means PID:5, source cluster:A
             5 CB ------------> F(99,5) ----> F(99,5)
                           5 ---CB   # Control Batch appended when A failover to B
              5(B) -------> F5(17,5B) # 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.

In B and C, even if 2 records with PID 5, they won't duplicate with each other because of the control batch.

The mirror partition state transitions are:

• STOPPING: remove fetchers, truncate to LSO, persist last mirrored offsets, write MIRROR_PID_RESET barrier
• STOPPED:  partition is writable (terminal state, no actions)

The key follows the standard control record format (version=0, type=7). The value uses the MirrorPidResetRecord schemaKey schema:

{
  "apiKey": 3,
  "type": "coordinator-keydata",
  "name": "MirrorPidCollisionKeyMirrorPidResetRecord",
  "validVersions": "0",
  "flexibleVersions": "none0+",
  "fields": [
    { "name": "TopicIdVersion", "type": "uuidint16", "versions": "0",
      "about": "The destination topic ID whereversion of the collision occurred."},
    { "name": "Partition", "type": "int32", "versions": "0",
      "about": "The partition indexmirror PID reset record."},
    { "name": "SourceClusterId", "type": "string", "versions": "0",
      "about": "The source cluster UUID whose producer ID was remappedfor verification."},
    { "name": "OriginalPid", "type": "int64", "versions": "0",
      "about": "The original producer ID on the source cluster before mapping."}
  ]
}

Value schema:

{
  "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."}
  ]
}

...

]
}

The SourceClusterId field records which source cluster the mirrored data came from, enabling future validation (e.g. detecting unexpected source cluster changes) and data provenance tracing from the log itself.

When the barrier batch is encountered during append or during log recovery, all producer entries are removed from the PSM. This ensures leaders, followers, and recovery all handle the barrier consistently. Because the partition is read-only during mirroring, all PSM entries originate from mirrored data. Expiring all entries is safe: no local producer state exists to preserve. Control batches are filtered out by the consumer fetcher via isControlBatch checks. The barrier is invisible to application consumers, just like transaction markers (commit/abort). The log dump tool is enhanced to deserialize and display MIRROR_PID_RESET records.

The barrier approach works correctly with all practical mirroring topologies:                                                                                                                                                                                                            

• Active-passive (A to B): B mirrors from A, stores records as-is. On failover, barrier expires all PSM entries. Local producers get fresh PIDs from the coordinator with no collision risk.                                                                                            

• Failback (A to B, then B to A): After failover, B becomes writable. Later, A starts mirroring from B and truncates its log to the LSO. A then stores B's records as-is. B's barrier record is included in the fetched data and appended to A's log, triggering PSM expiration on A. This is consistent with the general rule: when the barrier batch is encountered during append or during log recovery, all producer entries are removed from the PSM. A will write its own barrier when it eventually stops mirroring from B, producing a clean slate before A becomes writable again.

• Fan-out (A to B, A to C): B and C mirror independently from A, each with its own PSM per partition. On failover, each writes its own barrier independently.
• Fan-in (A to C, B to C, different topics): Each topic's partitions have independent PSMs. The barrier is written per partition during the STOPPING transition of each mirror.
• Chain (A to B to C): B mirrors from A, stores records as-is. C mirrors from B, stores records as-is. On failover at any point in the chain, the barrier expires all PSM entries on the stopping node. Longer chains work inductively by the same principle.