Page History

...

The approach is to proactively expire the stale producer state on failover. The key insight is that, during mirroring, the destination partition is read-only: no local producers exist, so all ProducerStateManager (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 record (type 7) is written to each destination partition's log during the STOPPING state transition, after the fetcher has been removed and truncation to LSO is completed, but before the partition becomes writable.

...

• STOPPING: Remove fetchers, truncate to LSO, persist LME, write MIRROR_PID_RESET control batchrecord.
• STOPPED: Partition is writable (terminal state, no actions).

...

When the control batch is encountered during append or during log recovery, all producer entries are removed from the PSM. This ensures both leaders and followers handle the control batch barrier consistently. Given that 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, just like transaction markers (commit/abort). The log dump tool is enhanced to deserialize MirrorPidResetRecordMIRROR_PID_RESET records.

The control record approach works correctly with all practical mirroring topologies:                                                                                                                                                                                                            

• Active-passive (A to B): B mirrors from A, stores records as-is. On failover, the MirrorPidResetRecord expires MIRROR_PID_RESET record 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 MirrorPidResetRecord MIRROR_PID_RESET 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 MIRROR_PID_RESET batch record is encountered during append or during log recovery, all producer entries are removed from the PSM. A will write its own MirrorPidResetRecord MIRROR_PID_RESET record 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 MirrorPidResetRecord MIRROR_PID_RESET record independently.
• Fan-in (A to C, B to C, different topics): Each topic's partitions have independent PSMs. The MirrorPidResetRecord is MIRROR_PID_RESET record 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 MirrorPidResetRecord expires MIRROR_PID_RESET record expires all PSM entries on the stopping node. Longer chains work inductively by the same principle.

...

1. The user sends CreateMirror requests to any broker with the mirror name and mirror related properties (bootstrap servers, security settings, etc.).
2. The broker forwards the request to the active controller.
3. The controller saves the properties into the metadata log as ConfigRecord entries with ConfigResource(Type.MIRROR, mirrorName)type MIRROR.
4. If this is the first mirror being created, the controller also auto creates the __mirror_state internal topic.
5. All brokers receive the metadata update and the MirrorMetadataManager registers the new mirror configuration.

...

1. User sends AddTopicsToMirror request with topics and mirror name.

2. The broker forwards to the active controller.

3. The controller validates that each topic exists and is not already in a mirror. It then sets the topic config mirror.name=<mirrorName> for each topic, generating a ConfigRecord per topic into the metadata log.
4. Response is sent back to clients with per topic results.
5. When the MirrorMetadataManager in the partition leader node gets notified about the topic config update, it detects that mirror.name is not empty and has no .removed or .paused suffix. It then queries the current mirror partition state from the coordinator. The coordinator could be located on a different broker node, so a ReadMirrorStates inter broker RPC may be needed.
6. Based on the current mirror partition state, the state machine transitions the partition. In most cases, from UNKNOWN to PREPARING.
7. During PREPARING, the mirror fetcher performs Last Mirrored Epoch (LME) truncation. The LME is the greatest leader epoch that the source cluster recognizes from the destination. If the source has no LME knowledge (first time mirroring), it returns -1 and the destination truncates everything and replicates from scratch. Otherwise, the destination truncates at the start offset of the first epoch beyond the LME. It then waits until all ISR members (or all replicas if mirror.support.unclean.leader.election=true) complete the truncation.
8. Once all ISR members have completed truncation, the state transitions from PREPARING to MIRRORING. A MirrorFetcherThread is created and starts sending consumer Fetch requests (not follower requests) to the source cluster to replicate data. The Fetch protocol handles any offset level divergence by truncating to the exact offset where the source epoch ends. The fetched batch retains its original leader epoch from the source. When the partition later becomes writable after failover, the leader epoch is bumped to ensure monotonically increasing epochs for new records.
9. The partition state is persisted to the __mirror_state topic on each state change via local append or WriteMirrorStates (when coordinator is remote) as MirrorPartitionStateKey/MirrorPartitionStateValue records, distributed by hash(mirrorName, topicId, partition) % numPartitions.
10. The MirrorMetadataManager also periodically synchronizes topic configs, consumer group offsets, and ACLs from the source cluster.

...

1. User sends RemoveTopicsFromMirror request with topics and mirror name.
2. The controller validates each topic belongs to the specified mirror and is in MIRRORING state. It then updates the topic config by appending the .removed suffix, e.g. mirror.name=cluster1my-mirror.removed, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects the .removed suffix on mirror.name. It queries the current mirror partition state from the coordinator, and transitions the mirror partition to STOPPING.
4. During STOPPING:
   1. The MirrorFetcherManager removes all fetcher threads for the affected partitions, stopping replication.
   2. Bump the leader epoch for the partitions to ensure monotonically increasing epochs for new records.
   3. The log is truncated to LSO for transactional consistency.
   4. The LME is recorded as LastMirroredEpochsKey/LastMirroredEpochsValue records into the __mirror_state topic for potential future failback.
   5. A MIRROR_PID_RESET control batch record is written to the partition log, which expires all ProducerStateManager entries so that new producers get fresh PIDs with no collision risk.
5. The state transitions from STOPPING to STOPPED. The read only flag is cleared and the topic becomes writable. New producers can start producing with fresh PIDs starting at sequence 0 and a higher leader epoch.

...

1. User sends PauseMirrorTopics request with topics and mirror name.
2. The controller validates each topic belongs to the specified mirror and is currently in MIRRORING state. It appends the .paused suffix to the mirror name config, e.g. mirror.name=cluster1my-mirror.paused, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects the .paused suffix. It transitions the state to PAUSING.
4. During PAUSING, the MirrorFetcherManager removes the fetcher threads for the affected partitions. No more data is replicated.
5. The state transitions from PAUSING to PAUSED. The partition remains read only. Metadata synchronization (configs, groups, ACLs) is also halted for the paused topics.
6. The partition state change is persisted to the __mirror_state topic.

...

1. User sends ResumeMirrorTopics request with topics and mirror name.
2. The controller validates the topic is currently paused (has .paused suffix). It removes the .paused suffix, restoring the original mirror name, e.g. mirror.name=cluster1, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects that mirror.name no longer has the .paused suffix.
4. The state transitions directly from PAUSED to MIRRORING. No log truncation is needed because the partition is already at the correct offset from before the pause.
5. New MirrorFetcherThread instances are created and resume replication from the current log end offset.
6. Metadata synchronization (configs, groups, ACLs) also resumes.

...

1. The user sends a DeleteMirror request with the mirror name.
2. The controller validates that the mirror is empty (no topics assigned) or all its partitions are in STOPPED state.
3. If valid, the controller tombstones the mirror configuration in the cluster metadata log, removing all ConfigRecord entries for the mirror.
4. All The mirror state entries records in __mirror_state for this mirror are cleaned upinternal topic are also tombstoned.
5. Any remaining coordinator state is shut down, source cluster connections are closed, and the mirror name becomes available for reuse.
6. After deletion, failback using this mirror configuration is no longer possible.

...