Page History

...

• State Management: Mirror configuration and partition states are stored in the internal topic. The coordinator loads the state on startup and partition leadership changes.
• Partition Assignment: Mirror partitions are distributed evenly to coordinators across all brokers and allows for horizontal scaling. The number of coordinator partitions is configurable via mirror.topic.num.partitions.
• Leader Election: When a broker becomes the leader for a __mirror_state partition, it loads the mirror metadata for all mirrors assigned to that partition and begins coordinating those mirrors. On resignation, it clears its in-memory state to avoid stale metadata.
• Metadata Refresh Scheduling: The coordinator schedules periodic metadata refresh operations by invoking a metadata manager every 30 seconds by default. This ensures that topology changes, configuration updates, and offset commits in the source cluster are continuously propagated to the destination cluster. The refresh interval is configurable via mirror.metadata.refresh.interval.ms.
• State Transitions: The coordinator manages asynchronous state transitions for mirror partitions. Each partition is an independent replication unit with its own state. When the coordinator is the leader for a mirror partition, it writes the state updates directly to the internal topic. Remote brokers read and write partition state via new RPCs, enabling distributed coordination across the cluster.

...

Image Added

Figure 3: Mirror Partition Lifecycle.

States descriptions:

• UNKNOWN: The partition has no cached state (broker just became leader, state not loaded yet). Not an explicit API-driven state, just the absence of state.
• PREPARING: The coordinator for this partition detects via onMetadataUpdate that it leads a mirror partition. It fetches last mirrored offsets from the source cluster and truncate truncates logs to align the local log with the source.
  Valid from: null, UNKNOWN, STOPPED, FAILED.
• MIRRORING: All ISR members have completed truncation. A MirrorFetcherThread is started to continuously replicate records from the source cluster. Valid from: PREPARING, PAUSED.
• PAUSING: Triggered by PauseMirrorTopics API (appends .paused suffix to mirror.name config). The system removes fetchers for the affected partitions. Valid from: MIRRORING only.
• PAUSED: Fetchers have been removed. The partition stays read-only with no active fetchers and no metadata sync (configs, consumer groups, ACLs). On resume, transitions directly to MIRRORING (fetchers resume from local LEO, no truncation needed). Valid from: PAUSING only.
• STOPPING: Triggered by RemoveTopicsFromMirror API (user wants to fail over) or topic deletion on the source. The system records the last mirrored offset to the internal topic. Valid from: PREPARING, MIRRORING, PAUSING (race guard, TODO: remove once shared queue serializes transitions), PAUSED.
• STOPPED: Last mirrored offsets have been persisted. The topic becomes writable on the destination cluster (the mirror fetcher is removed and the read-only flag is cleared). Valid from: STOPPING only.
• FAILED: An error occurred. Can transition back to to PREPARING to retry. Valid from: any state.
• (TODO: pause/resume)

Scenarios:Example scenarios:

• Starting a mirror (UNKNOWN -> PREPARING -> MIRRORING): The addTopicsToMirror command sets mirror.name config via the controller. The metadata update propagates to brokers. The broker leading the partition finds out the partition state via the coordinator, and this might trigger readMirrorState RPC to query from the remote coordinator and transitions to PREPARING if it's in a valid transition state (e.g. UNKNOWN). After truncation completes, it moves to MIRRORING and starts the mirror fetcher to fetch data from the source cluster.
• Failover (MIRRORING -> STOPPING -> STOPPED): The removeTopicsFromMirror command appends the ".removed” suffix to the mirror.name config. The partition leader  detects the stop request, transitions to STOPPING, persists the last offset, then moves to STOPPED. The topic is now writable after the STOPPED state.
• Restarting a stopped mirror (STOPPED -> PREPARING -> MIRRORING): The mirror.name config is set again. onMetadataUpdate sees the partition in STOPPED state and transitions to PREPARING, re-truncating and resuming replication.

MirrorMetadataManager

The MirrorMetadataManager (MMM) implements periodic metadata synchronization between source and destination clusters. It maintains persistent network connections to all source clusters.

...

Add a topic or set of topics to an existing cluster mirror (start mirroring; the topic flag accepts regex expression to operate on multiple topics at once):

$ bin/kafka-mirror.sh --bootstrap-server :9094 --add --topic my-topic --mirror my-mirror --replication-factor 2
Added 1 topic(s) to mirror my-mirror: [my-topic]

...

$ bin/kafka-mirrors.sh --bootstrap-server :9094 --describe
MIRROR                         TOPIC                                    PARTITION  SOURCE-OFFSET   DESTINATION-OFFSET LAG      STATE       
my-mirror                      bar                                      0          2324            2324               0        MIRRORING   
my-mirror                      foo                                      0          69              66                 3        MIRRORING   
my-mirror                      foo                                      1          94              84                 10       MIRRORING   
my-mirror                      foo                                      2          94              90                 4        MIRRORING   
new-mirror                     baz                                      0          189             189                0        MIRRORING   
new-mirror                     baz                                      1          859             859                0        MIRRORING

Pause /resume mirroring for a specific topic or set of topics (topics reamin remain read-only):

$ bin/kafka-mirrors.sh --bootstrap-server :9094 --pause --topic my-topic --mirror my-mirror
Paused mirroring for 1 topic(s): [my-topic]

Resume mirroring for a specific topic or set of topics:

TODO$ bin/kafka-mirrors.sh --bootstrap-server :9094 --resume --topic my-topic --mirror my-mirror
Resumed mirroring for 1 topic(s): [my-topic]

Remove a specific topic or set of topics from a mirror (failover: all partitions ; topics become writable):

$ bin/kafka-mirror.sh --bootstrap-server :9094 --remove --topic my-topic --mirror my-mirror
Removed 1 topic(s) from mirror my-mirror: [my-topic]

...