DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
RPC | Component | ACL Operation | ACL Resource | Purpose |
| Fetch | MFT | Read | Topic | Data replication |
| Metadata | MMM | Describe | Topic | Topic discovery and leader tracking |
| DescribeConfigs | MMM | DescribeConfigs | Topic | Topic configuration sync |
| ListGroups | MMM | Describe | Group | Consumer group offset sync |
| OffsetFetch | MMM | Describe | Group | Consumer group offset sync |
| DescribeAcls | MMM | Describe | Cluster | ACL synchronization |
| LastMirroredEpochsDescribeMirrors | MC | Read | Cluster | Log truncation when preparing |
| ApiVersions | MMM | Feature negotiation | ||
| ListOffsets | MFT | Describe | Topic | Offset bounds discovery |
| OffsetsForLeaderEpoch | MFT | Describe | Topic | Leader epoch validation for truncation |
...
- User sends RemoveTopicsFromMirror request with topics and mirror name.
- 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=my-mirror.removed, generating a ConfigRecord.
- When the MirrorMetadataManager 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.
- During STOPPING:
- The MirrorFetcherManager removes all fetcher threads for the affected partitions, stopping replication.
- Bump the leader epoch for the partitions to ensure monotonically increasing epochs for new records.
- The log is truncated to LSO for transactional consistency.
- The LME is recorded as LastMirroredEpochsKeysKey/LastMirroredEpochsValue LastMirrorEpochsValue records into the __mirror_state topic for potential future failback.
- 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.
- 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.
...
- The user sends DescribeMirrorsRequest with optional mirror names (empty means all mirrors).
The broker handler queries two sources:
The ReplicaManager which provides source offset, destination offset, and lag for each partition.
- The MirrorCoordinator which provides the current partition state from the metadata manager cache.
The request is forwarded to each broker that only reports partitions for which it has lag information or is the partition leader. This avoids duplicate reporting across brokers.
- For each partition, the response includes: mirror name, topic name, partition ID, source offset, destination offset, lag, and current state, and LME.
- No metadata records are written. This is a read only operation.
...
Returns the current mirroring status, state, and configuration for the specified mirror topics on the destination cluster. Allows destination cluster partition leaders to query the LME from the source cluster.
DescribeMirrorsRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker"],
"name": "DescribeMirrorsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorNames", "type": "[]string", "versions": "0+", "entityType": "mirrorName",
"about": "The names of the mirrors to describe. Null or empty array means all mirrors." },
{ "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+", "default": "false",
"about": "Whether to include authorized operations." }
]
} |
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "DescribeMirrorsResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." },
{ "name": "Mirrors", "type": "[]DescribedMirror", "versions": "0+",
"about": "Each described mirror.", "fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name." },
{ "name": "Topics", "type": "[]TopicPartitions", "versions": "0+",
"about": "Each topic in the mirror.", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionDetail", "versions": "0+",
"about": "Each partition detail.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "SourceOffset", "type": "int64", "versions": "0+", "default": "-1",
"about": "The high watermark offset from the source cluster leader, or -1 if not yet available." },
{ "name": "DestinationOffset", "type": "int64", "versions": "0+", "default": "-1",
"about": "The log end offset on the destination cluster, or -1 if not yet available." },
{ "name": "Lag", "type": "int64", "versions": "0+", "default": "-1",
"about": "The lag (source offset - destination offset), or -1 if not yet available." },
{ "name": "State", "type": "string", "versions": "0+",
"about": "The partition state." },
{ "name": "", "type": "int32", "versions": "0+", "default": "-1",
"about": "The last mirror leader epoch, or -1 if not available." }
]}
]}
]}
]
} |
ReadMirrorStates
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "ReadMirrorStatesResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." },
{ "name": "Topics", "type": "[]TopicResult", "versions": "0",
"about": "The read results for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
"about": "The results for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "LastMirroredOffsetLastMirrorEpoch", "type": "int64int32", "versions": "0", default": "-1",
"about": "The last mirrored offset mirror leader epoch, or -1 if not available." },
{ "name": "State", "type": "int8", "versions": "0+",
"about": "The mirror partition state." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "WriteMirrorStatesRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The mirror name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0",
"about": "The data for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData", "versions": "0",
"about": "The data for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "LastMirroredOffset", "type": "int64int32", "versions": "0", "default": "-1",
"about": "The last mirrored offset mirror leader epoch, or -1 if not available." },
{ "name": "State", "type": "int8", "versions": "0+",
"about": "The mirror partition state." }
]}
]},
{ "name": "RemovedTopics", "type": "[]string", "versions": "0+", "about": "The topic names to be removed." }
]
} |
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "WriteMirrorStatesResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." },
{ "name": "Topics", "type": "[]TopicResult", "versions": "0",
"about": "The write results for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
"about": "The results for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
...
BumpLeaderEpochs
Allows destination cluster partition leaders to query the LME from the source cluster.
LastMirroredEpochsRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "LastMirroredEpochsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0",
"about": "The data for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData", "versions": "0",
"about": "The data for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." }
]}
]}
]
} |
LastMirroredEpochsResponse
| Code Block |
|---|
{
"apiKey": 97,
"type": "response",
"name": "LastMirroredEpochsResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." },
{ "name": "Topics", "type": "[]TopicResult", "versions": "0",
"about": "The results for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
"about": "The results for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "LastMirroredEpoch", "type": "int32", "versions": "0", "default": "-1",
"about": "The last mirrored epoch, or -1 if not available." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
BumpLeaderEpochs
Allows destination cluster partition leaders to bump the leader epoch on the destination bump the leader epoch on the destination controller so that it is at least as high as the source cluster's leader epoch.
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "BumpLeaderEpochsResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "Topics", "type": "[]TopicPartitions", "versions": "0+",
"about": "Each topic in the mirror.", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionDetail", "versions": "0+",
"about": "Each partition state.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
Metadata Records
...
LastMirrorEpochs
The greatest leader epoch of a given partition that a destination cluster recognizes from the source cluster.
| Code Block |
|---|
{
"apiKey": 1,
"type": "coordinator-key",
"name": "LastMirroredEpochsKeysKey",
"validVersions": "0",
"flexibleVersions": "none",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0",
"about": "The cluster mirror name."}
]
}
{
"apiKey": 1,
"type": "coordinator-value",
"name": "LastMirroredEpochsValueLastMirrorEpochsValue",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Topics", "type": "[]Topic", "versions": "0+",
"about": "The mirror topics for which we want to store the last mirrored epochs.", "fields": [
{ "name": "Name", "type": "string", "versions": "0",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]Partition", "versions": "0+",
"about": "Each partition to record the last mirrored epochs.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "LastMirroredEpoch", "type": "int32", "versions": "0+",
"about": "The last mirror mirroredleader epoch for this partition." }
]}
]}
]
} |
...
Key | Description | Default |
|---|---|---|
mirror.name | Identifies the mirror that manages this topic. Topics with this configuration set are read-only and can only be modified through mirror management APIs. This property is filtered out from DescribeConfigs responses to avoid exposing internal state to users. | |
mirror.replication.throttled.replicas | A list of replicas for which log replication should be throttled on the mirror follower node. The list should describe a set of replicas in the form [PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... or alternatively the wildcard '*' can be used to throttle all replicas for this topic." | |
mirror.support.unclean.leader.election | When enabled, LastMirroredEpoch log LME log truncation waits for all replicas (not just ISR members) to join the ISR and complete the truncation. | false |
...
| Code | Name | Message | Used By |
|---|---|---|---|
| 3 | UNKNOWN_TOPIC_OR_PARTITION | The topic does not exist on the target cluster | RemoveTopicsFromMirror, PauseMirrorTopics, ResumeMirrorTopics |
| 15 | COORDINATOR_NOT_AVAILABLE | The mirror coordinator is not active | WriteMirrorStates, ReadMirrorStates, LastMirroredOffsets |
| 31 | CLUSTER_AUTHORIZATION_FAILED | The client is not authorized to perform the mirror operation | WriteMirrorStates, ReadMirrorStates, LastMirroredOffsets |
| 35 | UNSUPPORTED_VERSION | Cluster mirroring is disabled (mirror.version=0) | CreateMirror, AddTopicsToMirror, RemoveTopicsFromMirror, PauseMirrorTopics, ResumeMirrorTopics, ListMirrors, DescribeMirrors, DeleteMirror |
| TBD | MIRROR_AUTHORIZATION_FAILED | Mirror authorization failed | CreateMirror, AddTopicsToMirror, RemoveTopicsFromMirror, PauseMirrorTopics, ResumeMirrorTopics, DeleteMirror |
| TBD | READ_ONLY_TOPIC | The topic is read-only because it is a mirror topic on the target cluster | Produce |
| TBD | INVALID_MIRROR_NAME | The mirror name does not meet the naming rules | CreateMirror |
| TBD | UNKNOWN_MIRROR | The topic is not assigned to any mirror | RemoveTopicsFromMirror, PauseMirrorTopics, ResumeMirrorTopics |
| TBD | TOPIC_ALREADY_IN_MIRROR | The topic is already assigned to a mirror | AddTopicsToMirror |
| TBD | TOPIC_NOT_IN_MIRROR | The topic does not belong to the specified mirror | RemoveTopicsFromMirror, PauseMirrorTopics, ResumeMirrorTopics |
| TBD | MIRROR_TOPIC_ALREADY_PAUSED | The mirror topic is already paused | PauseMirrorTopics |
| TBD | MIRROR_TOPIC_NOT_PAUSED | The mirror topic is not paused | ResumeMirrorTopics |
| TBD | MIRROR_TOPIC_BEING_REMOVED | The mirror topic is being removed | ResumeMirrorTopics |
| TBD | MIRROR_NOT_EMPTY | The mirror still has active or non-removed topics | DeleteMirror |
...
| MirrorMaker 2 | Cluster Mirroring | |
|---|---|---|
Offset Translation | Lossy, requires remapping, causes reprocessing overhead | None needed, offsets preserved exactly |
Metadata Sync | Requires separate connector configuration (MirrorSourceConnector, MirrorCheckpointConnector) | Automatic (topics, configs, consumer groups, ACLs) |
Transactional Topics | Markers copied as regular records, incomplete transactions possible during replication | Markers mirrored, LSO truncation ensures consistency before failover |
Topic Write Protection | Not supported (mirror topics always writable) | Read-only enforcement during mirroring, writable only after explicit failover |
Tiered Storage | Fetches from broker (which reads from remote storage) | Not initially supported (future work) |
Active-Active | Supported via topic prefixing and cycle detection | Not supported (read-only enforcement prevents cycles) |
Share Groups | Not supported | Supported |
Failback | Full re-mirror from offset 0 | Delta sync using LastMirroredOffsetDescribeMirror |
Topic Name Preservation | No, destination topics prefixed with source cluster alias (e.g., source.topic-name) | Yes, same topic name as source |
Topic ID Preservation | No, destination gets new topic ID | Yes: same topic ID as source |
...