Page History

...

In this example, source cluster log at the time of failure:

Destination cluster log at failover (replication reached offset 2):

After the STOPPING transition appends abort markers: 

Transaction 4001 was committed at the source but aborted at the destination because the COMMIT marker (offset 3) had not yet been replicated. Transaction 4002 was correctly aborted at both clusters. Applications that require strict transactional guarantees across clusters should implement deduplication or reconciliation logic after failover. Additionally, the kafka-transactions tool can only abort transactions originated from the local cluster. It cannot abort transactions replicated via mirroring because the __transaction_state topic is not mirrored. Hanging transactions from mirrored data are resolved exclusively by the STOPPING transition flow described above.

...

Failover is initiated by calling the StopMirrorTopics API API, which appends a ".removed" suffix stopped suffix to the mirror.name internal config. This transitions the mirror topics from read-only to writable state after the stopping process completes gracefully. When producers reconnect to the destination cluster after failover, they obtain new producer IDs which are separate from previously mirrored IDs, so they begin writing with fresh sequence numbers starting from 0. Consumers can reconnect to the destination cluster using the same group ID, resuming from the last synchronized offsets, minimizing data re-processing or gaps. The transition is transparent from the consumer's perspective and offset management continues normally through the destination's group coordinator.

...

1. User sends StartMirrorTopics 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 stopped 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 Mirror 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.
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 StopMirrorTopics 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 stopped suffix, e.g. mirror.name=my-mirror.removedstopped, generating a ConfigRecord.
3. When the MirrorMetadataManager gets notified, it detects the .removed stopped suffix on mirror.name. It queries the current mirror partition state from the coordinator, and transitions the mirror partition to STOPPING.
4. During STOPPING, the following operations execute sequentially:
   
   1. The MirrorFetcherManager removes all fetcher threads for the affected partitions, stopping replication.
   2. The LME is recorded as LastMirrorEpochsKey/LastMirrorEpochsValue records into the __mirror_state topic for potential future failback.
   3. The leader epoch is bumped for the partitions to ensure monotonically increasing epochs for new records.

   4. ABORT markers are appended for all ongoing transactions. For each partition, ProducerStateManager provides the set of in-flight transaction entries and an EndTransactionMarker(ABORT) is appended for each one. This resolves hanging transactions without truncating committed data.

   5. A MIRROR_PID_RESET control 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.

...

Create a new cluster mirror in the destination cluster (forbidden suffixes: .removedstopped, .paused):

$ echo "bootstrap.servers=localhost:9092" >/tmp/mirror.properties
$ bin/kafka-mirror.sh --bootstrap-server :9094 --create --mirror my-mirror --mirror-config /tmp/mirror.properties
Created mirror my-mirror

...

Allows users to create a mirror and supply its configuration. When the broker receives the request, it validates that the mirror name is not already in use, contains only permitted characters, and does not end with "with .removed" stopped or " .paused " suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.

...

Stop mirroring for the specified topics. The broker validates that all target topic partitions are in either PREPARING or MIRRORING state. Once validated, the request is forwarded to the controller, which appends the ".removed" suffix the .stopped suffix to the mirror.name topic config to mark the topics as no longer mirrored.

...