A producer ID (PID) is a 64-bits long that has the following rules: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
4. Identity: A PID identifies exactly one logical idempotent producer within a partition's log.
5. Stability: A logical idempotent producer always maps to the same PID, regardless of mirror topology changes or broker restarts.
6. Positivity: Local PIDs are non-negative, allocated sequentially from 0. PID -1 means non-idempotent producer.

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.

...

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

...

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: bit-field mapping with collision detection

Apply 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:

...

  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.

The probability of a collision is negligible but, when it happens, we propose a resolution strategy to resolve it. We introduce the sourceClusterId optional field in ProducerSnapshot.json schema, bumping the version to 2. The sourceClusterId will be set only for mirrored PIDs, null for local producers.

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.

Key schema:

{
  "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",
      

{
  "name": "SourceClusterId",
  "type": "string",
  "versions": "2+",
  "nullableVersions": "2+",
  "default": "null",
  "about": "The source cluster UUID whose producer ID for mirrored producer IDs, null for local producers"
} was remapped."},
    { "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."}
  ]
}

On leader election, the coordinator loads collision records from __mirror_state  and populates the collision cache. When mapping a PID, checks this cache first. If a known salt exists, it is applied directly without re-probing the origins map. A null value (tombstone) deletes the record when the corresponding producer state expiresCollision detection checks the PSM cache for existing entries with a different source cluster ID, rehashing with an incrementing salt on conflict and forcing a PSM cache snapshot.