DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
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 record (type 7) is written to each destination partition's log during the STOPPING state transition, just before the partition becomes writable.
The key follows When the standard control record format (version=0, type=7). The value uses the following schema:
| Code Block |
|---|
{
"type": "data",
"name": "MirrorPidResetRecord",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Version", "type": "int16", "versions": "0",
"about": "The version of the mirror PID reset record."},
{ "name": "SourceClusterId", "type": "string", "versions": "0",
"about": "The source cluster UUID for verification."}
]
} |
The SourceClusterId field records which source cluster the mirrored data came from, enabling future validation (e.g. detecting unexpected source cluster changes) and data provenance tracing from the log itself.
...
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 MIRROR_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 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 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 record is encountered during append or during log recovery, all producer entries are removed from the PSM.
...
- A will write its own 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 MIRROR_PID_RESET record independently.
- Fan-in (A to C, B to C, different topics): Each topic's partitions have independent PSMs. The 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 MIRROR_PID_RESET record expires all PSM entries on the stopping node. Longer chains work inductively by the same principle.
Exactly-Once Semantics
Cluster mirroring does not support exactly-once semantics across clusters. Transactional support means that after failover, the destination cluster will not have hanging transactions that block READ_COMMITTED consumers, but it does not guarantee that committed records from the source are atomically synced to the destination.
The mirror fetcher thread uses READ_UNCOMMITTED isolation, so records from uncommitted transactions are replicated to the destination before the source decides them. This reduces replication lag compared to READ_COMMITTED, but means uncommitted data is visible to READ_UNCOMMITTED consumers on the destination before failover. When stop mirroring is triggered, ongoing transactions are decided by appending explicit ABORT markers, preserving all previously committed data.
In this example, source cluster log at the time of failure:
Offset | Type | PID | Content |
0 | DATA | 4001 | key=A, value=1 |
1 | DATA | 4001 | key=B, value=2 |
2 | DATA | 4002 | key=X, value=9 |
3 | COMMIT | 4001 | |
4 | DATA | 4003 | key=Y, value=5 |
5 | DATA | none | key=Z, value=10 |
Destination cluster log at failover (replication reached offset 2):
Offset | Type | PID | Content |
0 | DATA | 4001 | key=A, value=1 |
1 | DATA | 4001 | key=B, value=2 |
2 | DATA | 4002 | key=X, value=9 |
After the STOPPING transition appends abort markers:
...
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 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 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 record is encountered during append or during log recovery, all producer entries are removed from the PSM. A will write its own 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 MIRROR_PID_RESET record independently.
- Fan-in (A to C, B to C, different topics): Each topic's partitions have independent PSMs. The 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 MIRROR_PID_RESET record expires all PSM entries on the stopping node. Longer chains work inductively by the same principle.
Exactly-Once Semantics
Cluster mirroring does not support exactly-once semantics across clusters. Transactional support means that after failover, the destination cluster will not have hanging transactions that block READ_COMMITTED consumers, but it does not guarantee that committed records from the source are atomically synced to the destination.
The mirror fetcher thread uses READ_UNCOMMITTED isolation, so records from uncommitted transactions are replicated to the destination before the source decides them. This reduces replication lag compared to READ_COMMITTED, but means uncommitted data is visible to READ_UNCOMMITTED consumers on the destination before failover. When stop mirroring is triggered, ongoing transactions are decided by appending explicit ABORT markers, preserving all previously committed data.
In this example, source cluster log at the time of failure:
Offset | Type | PID | Content |
0 | DATA | 4001 | key=A, value=1 |
1 | DATA | 4001 | key=B, value=2 |
2 | DATA | 4002 | key=X, value=9 |
3 | COMMITABORT | 4001 | |
4 | DATAABORT | 4003 | key=Y, value=5 |
5 | DATA | none | key=Z, value=10 |
Destination cluster log at failover (replication reached offset 2):
Offset | Type | PID | Content |
0 | DATA | 4001 | key=A, value=1 |
1 | DATA | 4001 | key=B, value=2 |
2 | DATA | 4002 | key=X, value=9 |
After the STOPPING transition appends abort markers:
...
Offset
...
Type
...
PID
...
Content
...
0
...
DATA
...
4001
...
key=A, value=1
...
1
...
DATA
...
4001
...
key=B, value=2
...
2
...
DATA
...
4002
...
key=X, value=9
4002 |
Transaction 4001
...
3
...
ABORT
...
4001
...
4
...
ABORT
...
4002
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.
...
- 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, current state, and LME.
- No metadata records are written. This is a read only operation.
Public Interfaces
Command-Line
A new dump flag allows to decode cluster mirroring metadata for debugging purpose:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-dump-log.sh --mirror-state-decoder --files /tmp/server*/data/__mirror_state-1/00000000000000000000.log
Dumping /home/fvaleri/Documents/kafka/build/test/server4/data/__mirror_state-0/00000000000000000000.log
Log starting offset: 0
baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 0 CreateTime: 1771239071191 size: 98 magic: 2 compresscodec: none crc: 3314855113 isvalid: true
| offset: 0 CreateTime: 1771239071191 keySize: 13 valueSize: 17 sequence: -1 headerKeys: [] key: {"type":"2","data":{"mirrorName":"my-mirror"}} payload: {"version":"0","data":{"topicName":"my-topic","partition":0,"state":0}}
baseOffset: 1 lastOffset: 1 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 98 CreateTime: 1771239071219 size: 98 magic: 2 compresscodec: none crc: 3968746657 isvalid: true
| offset: 1 CreateTime: 1771239071219 keySize: 13 valueSize: 17 sequence: -1 headerKeys: [] key: {"type":"2","data":{"mirrorName":"my-mirror"}} payload: {"version":"0","data":{"topicName":"my-topic","partition":0,"state":1}} |
A new command-line tool kafka-mirrors.sh provides administrative operations for managing cluster mirrors:
- offset, lag, current state, and LME.
- No metadata records are written. This is a read only operation.
Public Interfaces
Command-Line
A new dump flag allows to decode cluster mirroring metadata for debugging purpose:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-dump-log.sh --mirror-state-decoder --files /tmp/server*/data/__mirror_state-1/00000000000000000000.log
Dumping /home/fvaleri/Documents/kafka/build/test/server4/data/__mirror_state-0/00000000000000000000.log
Log starting offset: 0
baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 0 CreateTime: 1771239071191 size: 98 magic: 2 compresscodec: none crc: 3314855113 isvalid: true
| offset: 0 CreateTime: 1771239071191 keySize: 13 valueSize: 17 sequence: -1 headerKeys: [] key: {"type":"2","data":{"mirrorName":"my-mirror"}} payload: {"version":"0","data":{"topicName":"my-topic","partition":0,"state":0}}
baseOffset: 1 lastOffset: 1 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 98 CreateTime: 1771239071219 size: 98 magic: 2 compresscodec: none crc: 3968746657 isvalid: true
| offset: 1 CreateTime: 1771239071219 keySize: 13 valueSize: 17 sequence: -1 headerKeys: [] key: {"type":"2","data":{"mirrorName":"my-mirror"}} payload: {"version":"0","data":{"topicName":"my-topic","partition":0,"state":1}} |
A new command-line tool kafka-mirrors.sh provides administrative operations for managing cluster mirrors:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --help
Create cluster mirrors and manage mirrored topics.
Option | ||
| Code Block | ||
| ||
$ bin/kafka-mirrors.sh --help Create cluster mirrors and manage mirrored topics. Option Description ------ ----------- --bootstrap-server <String: server to REQUIRED: The destination Kafka server connect to> to connect to. --command-config <String: command Property file containing configs to be config property file> passed to Admin Client. --create Create a new cluster mirror from a source cluster. --delete Delete a cluster mirror. --describe Describe a cluster mirror including partition lag and state. --exclude <String: exclude patterns> Comma-separated list of topic names or regex patterns to exclude from mirroring. Only valid with --start. --help Print usage information. --json Output description in JSON format --list Description List all cluster mirrors. ------ --mirror <String: mirror> The name of the cluster mirror. ----------mirror-config <String: mirror config Property file containing source property file> cluster configs for mirroring. --pause --bootstrap-server <String: server to REQUIRED: The destination Kafka server connect to> Pause mirroring for topics matching to connect to. --command-config <String: command Property file containing configs to be theconfig givenproperty patterns.file> --resume passed to Admin Client. --create Resume mirroring for previously paused Create a new cluster mirror from a topics matching the given patterns. --start source cluster. Start mirroring topics matching the --delete Delete a cluster mirror. given patterns. --describe --stop Describe a cluster mirror including Stop mirroring topics matching the partition lag and state. --exclude <String: exclude patterns> given patterns. Comma-separated list of topic names or --topics <String: topics> Comma-separated list of topic names or regex patterns to exclude from regex patterns (e.g., 'my-topic, mirroring. Only valid with --start. --help orders-.*,payments'). Print usage information. --version --json Display Kafka version. |
Create a new cluster mirror in the destination cluster (forbidden suffixes: .stopped, .paused):
| Code Block | ||
|---|---|---|
| ||
$ 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 |
Start mirroring a topic or set of topics (the topic flag accepts regex expression):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --start \ Output description in JSON format --list List all cluster mirrors. --mirror <String: mirror> The name of the cluster mirror. --mirror-config <String: mirror config Property file containing source property file> cluster configs for mirroring. --pause Pause mirroring for topics matching the given patterns. --topics 'orders-.*' --exclude 'orders-internal' --mirror my-mirror Started 2 mirror topic(s) in mirror my-mirror: [orders-us, orders-eu] |
Stop mirroring a topic or set of topics (failover; topics become writable):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --stop --topics 'orders-us' --mirror my-mirror
Stopped mirroring for 1 topic(s) in mirror my-mirror: [orders-us] |
Delete a mirror including its configuration (the mirror must be empty or include only stopped partitions):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --delete --mirror my-mirror
Deleted mirror my-mirror |
Pause mirroring for a specific topic or set of topics (topics remain read-only):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --pause --topic my-topic --mirror my-mirror
Paused mirroring for 1 topic(s) in mirror my-mirror: [my-topic] |
Resume mirroring for a specific topic or set of topics:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --resume --topic my-topic --mirror my-mirror
Resumed mirroring for 1 topic(s) in mirror my-mirror: [my-topic] |
...
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --list MIRRORresume Resume mirroring for previously paused topics matching the given patterns. --start Start mirroring topics matching the TOPICS CLUSTER-ID BOOTSTRAP-SERVER my-mirror given patterns. 2 lBq12jYZRp-9wF3M9MPopg localhost:9091,localhost:9092 new-mirror --stop 1 lBq12jYZRp-9wF3M9MPopg Stop localhost:9091,localhost:9092 |
Describe configured mirrors to check their lag compared to their source topics (use --mirror flag to filter other partitions):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --describe MIRRORmirroring topics matching the TOPIC given patterns. PARTITION SOURCE-OFFSET --topics <String: topics> DESTINATION-OFFSET LAG Comma-separated list of topic names or STATE my-mirror bar regex patterns (e.g., 'my-topic, 0 - orders-.*,payments'). - --version - STOPPED my-mirror Display Kafka version. |
Create a new cluster mirror in the destination cluster (forbidden suffixes: .stopped, .paused):
| Code Block | ||
|---|---|---|
| ||
$ 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 |
Start mirroring a topic or set of topics (the topic flag accepts regex expression):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --start \ foo 0 69 66 3 MIRRORING my-mirror foo 1 94 84 10 MIRRORING my-mirror foo --topics 'orders-.*' --exclude 'orders-internal' --mirror my-mirror Started 2 mirror topic(s) in mirror my-mirror: [orders-us, orders-eu] |
Stop mirroring a topic or set of topics (failover; topics become writable):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --stop --topics 'orders-us' --mirror my-mirror Stopped mirroring 2for 1 topic(s) in mirror my-mirror: [orders-us] |
Delete a mirror including its configuration (the mirror must be empty or include only stopped partitions):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --delete --mirror my-mirror Deleted mirror my-mirror |
Pause mirroring for a specific topic or set of topics (topics remain read-only):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --pause --topic my-topic --mirror my-mirror
Paused mirroring for 1 topic(s) in mirror my-mirror: [my-topic] |
Resume mirroring for a specific topic or set of topics:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --resume --topic my-topic --mirror my-mirror
Resumed mirroring for 1 topic(s) in mirror my-mirror: [my-topic] |
List configured mirrors with additional information:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --list MIRROR94 90 4 MIRRORING new-mirror baz TOPICS CLUSTER-ID BOOTSTRAP-SERVER my-mirror 0 - 2 - lBq12jYZRp-9wF3M9MPopg localhost:9091,localhost:9092 new-mirror - PAUSED new-mirror1 lBq12jYZRp-9wF3M9MPopg localhost:9091,localhost:9092 |
Describe configured mirrors to check their lag compared to their source topics (use --mirror flag to filter other partitions):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --describe MIRROR baz TOPIC 1 - - PARTITION SOURCE-OFFSET DESTINATION-OFFSET LAG - PAUSED |
Alter mirror configuration (any valid configuration triggers a reconnection):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type mirrors --entity-name STATE my-mirror \ --alter --add-config 'bootstrap.servers=localhost:9092' Completed updating config for mirror my-mirror. |
Throttling on the destination cluster:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type brokers --entity-name 4 \ --alter --add-config mirror.replication.throttled.rate=100000000 Completed updating configbar for broker 4. $ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type topics --entity-name my-topic \ --alter --add-config mirror.replication.throttled.replicas=[0:4] Completed updating config for topic my-topic. |
Throttling on the source cluster:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-configs.sh --bootstrap-server :9091 --alter --add-config 'consumer_byte_rate=1024' \ --entity-type clients --entity-name broker-4-fetcher-0-mirror-my-mirror Completed updating config for client broker-4-fetcher-0-mirror-my-mirror. |
Grant mirror admin full access to a specific mirror:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
--cluster-mirror my-mirror \
--operation Create --operation Alter --operation Describe --operation Delete \
--operation AlterConfigs --operation DescribeConfigs \
--allow-principal User:mirror-admin
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`:
(principal=User:mirror-admin, host=*, operation=CREATE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=ALTER, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DELETE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) |
Grant read-only monitoring access to all mirrors:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
--cluster-mirror '*' \
--operation Describe --operation DescribeConfigs \
--allow-principal User:monitor
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=*, patternType=LITERAL)`:
(principal=User:monitor, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:monitor, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) |
List ACLs for a specific mirror:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-acls.sh --bootstrap-server :9094 --list --cluster-mirror my-mirror Current ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`: 0 - - - STOPPED my-mirror foo 0 69 66 3 MIRRORING my-mirror foo 1 94 84 10 MIRRORING my-mirror foo 2 94 (principal=User:mirror-admin, host=*, operation=CREATE, permissionType=ALLOW) 90 4 MIRRORING new-mirror baz 0 - - - PAUSED new-mirror baz 1 (principal=User:mirror-admin, host=*, operation=ALTER, permissionType=ALLOW) - - - PAUSED |
Alter mirror configuration (any valid configuration triggers a reconnection):
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type mirrors --entity-name my-mirror \
--alter --add-config 'bootstrap.servers=localhost:9092'
Completed updating config for mirror my-mirror. |
Throttling on the destination cluster:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type brokers --entity-name 4 \
--alter --add-config mirror.replication.throttled.rate=100000000
Completed updating config for broker 4.
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type topics --entity-name my-topic \
--alter --add-config mirror.replication.throttled.replicas=[0:4]
Completed updating config for topic my-topic. |
Throttling on the source cluster:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-configs.sh --bootstrap-server :9091 --alter --add-config 'consumer_byte_rate=1024' \
--entity-type clients --entity-name broker-4-fetcher-0-mirror-my-mirror
Completed updating config for client broker-4-fetcher-0-mirror-my-mirror. |
Grant mirror admin full access to a specific mirror:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
--cluster-mirror my-mirror \
--operation Create --operation Alter --operation Describe --operation Delete \
--operation AlterConfigs --operation DescribeConfigs \
--allow-principal User:mirror-admin
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`:
(principal=User:mirror-admin, host=*, operation=CREATE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=ALTER, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DELETE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) |
Grant read-only monitoring access to all mirrors:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-acls.sh --bootstrap-server :9094 --add \ --cluster-mirror '*' \ --operation Describe --operation DescribeConfigs \ --allow-principal User:monitor Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=*, patternType=LITERAL)`: (principal=User:mirror-adminmonitor, host=*, operation=DESCRIBE, permissionType=ALLOW) (principal=User:monitor, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) |
List ACLs for a specific mirror:
| Code Block | ||
|---|---|---|
| ||
$ bin/kafka-acls.sh --bootstrap-server :9094 --list --cluster-mirror my-mirror Current ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`: (principal=User:mirror-admin, host=*, operation=DELETECREATE, permissionType=ALLOW) (principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW) (principal=User:mirror-admin, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) |
Admin Client
New methods are added to the Admin interface for programmatic cluster mirror management, along with their supporting classes:
| Code Block | ||
|---|---|---|
| ||
/** * Create a new cluster mirror. * * @param mirrorName The name of the cluster mirror * @param configs Configuration for the cluster mirror, including bootstrap servers and security settings * @param options Options for the create mirror operation * @return The CreateMirrorResult */ CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options);/** * Create a new cluster mirror. * * @param mirrorName The name of the cluster mirror * @param configs Configuration for the cluster mirror, including bootstrap servers and security settings * @param options Options for the create mirror operation * @return The CreateMirrorResult */ CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options); /** * Start mirroring for the specified topics. * * When topics are started in a mirror, they become read-only on the destination cluster and start * replicating data from the source cluster. This operation marks the specified topics with the * mirror name, preventing local writes and enabling the MirrorFetcherThread to begin replication. * * @param mirrorName The cluster mirror name * @param topics Set of topic names to start mirroring * @param options Options for the start mirror topics operation * @return The StartMirrorTopicsResult containing futures for each topic */ StartMirrorTopicsResult startMirrorTopics(String mirrorName, Set<String> topics, StartMirrorTopicsOptions options); /** * Options for {@link Admin#startMirrorTopics(String, Set, StartMirrorTopicsOptions)}. */ public class StartMirrorTopicsOptions extends AbstractOptions<StartMirrorTopicsOptions> { private List<String> includePatterns = List.of(); private List<String> excludePatterns = List.of(); private Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata = Map.of(); public StartMirrorTopicsOptions includePatterns(List<String> patterns) { (principal=User:mirror-admin, host=*, operation=DELETE, permissionType=ALLOW) this.includePatterns = patterns; return this; } public StartMirrorTopicsOptions excludePatterns(List<String> patterns) { this.excludePatterns = patterns; return this; } public StartMirrorTopicsOptions topicMetadata(Map<String, StartMirrorTopicsRequestData.TopicData> metadata) { this.topicMetadata = metadata; return this; } public List<String> includePatterns() { return includePatterns; } public List<String> excludePatterns() { (principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW) return excludePatterns; } public Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata() { return topicMetadata; } } /** * Stop mirroring for the specified topics. * * This operation is typically used during failover scenarios when the destination cluster needs to * be promoted from passive (read-only mirror) to active (accepting writes). Stopping mirror topics * clears the mirrorName field from partition metadata, which allows producers to write * to these partitions. * * @param mirrorName The cluster mirror name * @param topics Set of topic names to stop mirroring * @param options Options for the stop mirror topics operation * @return The StopMirrorTopicsResult containing futures for each topic */ StopMirrorTopicsResult stopMirrorTopics(String mirrorName, Set<String> topics, StopMirrorTopicsOptions options); /** * Options for {@link Admin#stopMirrorTopics(String, Set, StopMirrorTopicsOptions)}. */ public class StopMirrorTopicsOptions extends AbstractOptions<StopMirrorTopicsOptions> { private List<String> patterns = List.of(); public StopMirrorTopicsOptions patterns(List<String> patterns) { this.patterns = patterns; return this; } public List<String> patterns() { (principal=User:mirror-admin, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) |
Admin Client
New methods are added to the Admin interface for programmatic cluster mirror management, along with their supporting classes:
| Code Block | ||
|---|---|---|
| ||
return patterns; } } /** * PauseCreate mirroringa fornew thecluster specified topicsmirror. * * Paused topics remain read-only on the destination cluster but stop fetching new data from the * source cluster. The mirror fetcher threads are removed for these partitions, preserving the * current replicated state. Mirroring can be resumed later with {@link #resumeMirrorTopics}@param mirrorName The name of the cluster mirror * @param configs Configuration for the cluster mirror, including bootstrap servers and security settings * @param options Options for the create mirror operation * @return The CreateMirrorResult */ CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options);/** * Create a new cluster mirror. * * @param mirrorName The name of the cluster mirror name * @param topics Set of topic names to pause mirroringconfigs Configuration for the cluster mirror, including bootstrap servers and security settings * @param options Options for the pausecreate mirror topics operation * @return The PauseMirrorTopicsResult containing futures for each topicCreateMirrorResult */ PauseMirrorTopicsResultCreateMirrorResult pauseMirrorTopicscreateMirror(String mirrorName, Set<String> topicsMap<String, String> configs, PauseMirrorTopicsOptionsCreateMirrorOptions options); /** * ResumeStart mirroring for previously paused topics. * * Resumed topics restart fetching the specified topics. * * When topics are started in a mirror, they become read-only on the destination cluster and start * replicating data from the source cluster,. pickingThis upoperation frommarks wherethe they specified *topics leftwith off.the New* mirror fetchername, threadspreventing arelocal createdwrites and enabling the partitions transitionMirrorFetcherThread back to the * MIRRORING statebegin replication. * * @param mirrorName The cluster mirror name * @param topics Set of topic names to resumestart mirroring * @param options Options for the resumestart mirror topics operation * @return The ResumeMirrorTopicsResultStartMirrorTopicsResult containing futures for each topic */ ResumeMirrorTopicsResultStartMirrorTopicsResult resumeMirrorTopicsstartMirrorTopics(String mirrorName, Set<String> topics, ResumeMirrorTopicsOptionsStartMirrorTopicsOptions options); /** * DeleteOptions afor cluster mirror including its configuration{@link Admin#startMirrorTopics(String, Set, StartMirrorTopicsOptions)}. */ public *class StartMirrorTopicsOptions Theextends mirrorAbstractOptions<StartMirrorTopicsOptions> must{ be empty (no topics)private orList<String> allincludePatterns its topics must have been removed (in STOPPED * state). After deletion, all mirror metadata are tombstoned and failback is no longer possible. * * @param mirrorName The cluster mirror name * @param options Options for the delete mirror operation * @return The DeleteMirrorResult */ DeleteMirrorResult deleteMirror(String mirrorName, DeleteMirrorOptions options); /** * List the cluster mirrors available in the cluster. * * @param options The options to use when listing the mirrors. * @return The ListMirrorsResult. */ ListMirrorsResult listMirrors(ListMirrorsOptions options); /** * Describe cluster mirrors. * * This operation retrieves detailed information about cluster mirrors including: * - Topics being mirrored * - Partition-level lag information (source offset vs destination offset) * - Mirroring state for each partition (INITIALIZING, PREPARING, MIRRORING, etc.) * * @param mirrorNames The names of the mirrors to describe * @param options The options to use when describing mirrors * @return The DescribeMirrorsResult */ DescribeMirrorsResult describeMirrors(Collection<String> mirrorNames, DescribeMirrorsOptions options); |
Protocol Changes
This section describes all protocol level changes.
CreateTopic
The CreateTopic API request is extended to add information required for mirror topic creation.
Request
| Code Block |
|---|
{ "name": "MirrorInfo", "type": "MirrorInfo", "versions": "8+", "nullableVersions": "8+", "ignorable": true,
"about": "Mirror information for creating a mirror topic from a source cluster.", "fields": [
{ "name": "TopicId", "type": "uuid", "versions": "8+",
"about": "The topic ID from the source cluster." }
]} |
The topic ID field ensures mirror topics retain the same topic ID as the source cluster topic. This allows fetch requests to pass validation on the source broker, and enables the system to verify that a topic being mirrored to a same-named topic in the destination cluster is indeed the same logical topic, not a name collision.
In normal topic creation, the MirrorInfo field will be null. When receiving the CreateTopic request, the controller will check the new field. If it is not set, the topic ID will be generated with random UUID as usual. Otherwise, the controller will do the following validation:
- This topic ID is not used by other topics in the current cluster
- The replicas for the partition assignment are all active and not in fenced or controlled shutdown. This is to make sure when a topic gets deleted and re-created with the same topic ID, the stale offline log dir won’t be treated as the active log dir after it becomes online (KAFKA-16234).
Fetch
The Fetch API is extended to add the MirrorLeaderEpoch field used by the destination cluster internal replication.
Request
| Code Block |
|---|
// new added field in in FetchPartition type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 2, "ignorable": true,
"about": "The latest known mirror leader epoch." }
|
Response
| Code Block |
|---|
// new added field in PartitionData type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 3, "ignorable": true,
"about": "The latest known mirror leader epoch." }, |
CreateMirror
Allows users to create a mirror and supply its configuration. The broker validates that the mirror name is not already in use, contains only permitted characters, and does not end with .stopped or .paused suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.
Request
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "CreateMirrorRequest",
// 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": "Config", "type": "[]MirrorConfig", "versions": "0+",
"about": "The cluster mirror configurations.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
"about": "The configuration key name." },
{ "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+",
"about": "The value to set for the configuration key."}
]}
]
} |
Response
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "CreateMirrorResponse",
// 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+",
"about": "The error message, or null if there was no error." }
]
} |
StartMirrorTopics
Start mirroring for the specified topics. The broker validates that all target topic partitions are in either UNKNOWN or STOPPED state; otherwise, the request is rejected. Once validated, the request is forwarded to the controller, which sets the mirror.name topic config to the specified mirror name.
Request
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "StartMirrorTopicsRequest",
// 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": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
{ "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The topic name." },
{ "name": "NumPartitions", "type": "int32", "versions": "0+",
"about": "The number of partitions for the topic. Must match the source topic." }
]},
{ "name": "IncludePatterns", "type": "[]string", "versions": "0+",
"about": "Regex patterns to add to mirror.topics.include." },
{ "name": "ExcludePatterns", "type": "[]string", "versions": "0+",
"about": "Regex patterns to add to mirror.topics.exclude." }
]
} |
Response
= List.of();
private List<String> excludePatterns = List.of();
private Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata = Map.of();
public StartMirrorTopicsOptions includePatterns(List<String> patterns) {
this.includePatterns = patterns;
return this;
}
public StartMirrorTopicsOptions excludePatterns(List<String> patterns) {
this.excludePatterns = patterns;
return this;
}
public StartMirrorTopicsOptions topicMetadata(Map<String, StartMirrorTopicsRequestData.TopicData> metadata) {
this.topicMetadata = metadata;
return this;
}
public List<String> includePatterns() {
return includePatterns;
}
public List<String> excludePatterns() {
return excludePatterns;
}
public Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata() {
return topicMetadata;
}
}
/**
* Stop mirroring for the specified topics.
*
* This operation is typically used during failover scenarios when the destination cluster needs to
* be promoted from passive (read-only mirror) to active (accepting writes). Stopping mirror topics
* clears the mirrorName field from partition metadata, which allows producers to write
* to these partitions.
*
* @param mirrorName The cluster mirror name
* @param topics Set of topic names to stop mirroring
* @param options Options for the stop mirror topics operation
* @return The StopMirrorTopicsResult containing futures for each topic
*/
StopMirrorTopicsResult stopMirrorTopics(String mirrorName, Set<String> topics, StopMirrorTopicsOptions options);
/**
* Options for {@link Admin#stopMirrorTopics(String, Set, StopMirrorTopicsOptions)}.
*/
public class StopMirrorTopicsOptions extends AbstractOptions<StopMirrorTopicsOptions> {
private List<String> patterns = List.of();
public StopMirrorTopicsOptions patterns(List<String> patterns) {
this.patterns = patterns;
return this;
}
public List<String> patterns() {
return patterns;
}
}
/**
* Pause mirroring for the specified topics.
*
* Paused topics remain read-only on the destination cluster but stop fetching new data from the
* source cluster. The mirror fetcher threads are removed for these partitions, preserving the
* current replicated state. Mirroring can be resumed later with {@link #resumeMirrorTopics}.
*
* @param mirrorName The cluster mirror name
* @param topics Set of topic names to pause mirroring
* @param options Options for the pause mirror topics operation
* @return The PauseMirrorTopicsResult containing futures for each topic
*/
PauseMirrorTopicsResult pauseMirrorTopics(String mirrorName, Set<String> topics, PauseMirrorTopicsOptions options);
/**
* Resume mirroring for previously paused topics.
*
* Resumed topics restart fetching data from the source cluster, picking up from where they
* left off. New mirror fetcher threads are created and the partitions transition back to the
* MIRRORING state.
*
* @param mirrorName The cluster mirror name
* @param topics Set of topic names to resume mirroring
* @param options Options for the resume mirror topics operation
* @return The ResumeMirrorTopicsResult containing futures for each topic
*/
ResumeMirrorTopicsResult resumeMirrorTopics(String mirrorName, Set<String> topics, ResumeMirrorTopicsOptions options);
/**
* Delete a cluster mirror including its configuration.
*
* The mirror must be empty (no topics) or all its topics must have been removed (in STOPPED
* state). After deletion, all mirror metadata are tombstoned and failback is no longer possible.
*
* @param mirrorName The cluster mirror name
* @param options Options for the delete mirror operation
* @return The DeleteMirrorResult
*/
DeleteMirrorResult deleteMirror(String mirrorName, DeleteMirrorOptions options);
/**
* List the cluster mirrors available in the cluster.
*
* @param options The options to use when listing the mirrors.
* @return The ListMirrorsResult.
*/
ListMirrorsResult listMirrors(ListMirrorsOptions options);
/**
* Describe cluster mirrors.
*
* This operation retrieves detailed information about cluster mirrors including:
* - Topics being mirrored
* - Partition-level lag information (source offset vs destination offset)
* - Mirroring state for each partition (INITIALIZING, PREPARING, MIRRORING, etc.)
*
* @param mirrorNames The names of the mirrors to describe
* @param options The options to use when describing mirrors
* @return The DescribeMirrorsResult
*/
DescribeMirrorsResult describeMirrors(Collection<String> mirrorNames, DescribeMirrorsOptions options); |
Protocol Changes
This section describes all protocol level changes.
CreateTopic
The CreateTopic API request is extended to add information required for mirror topic creation.
| Code Block |
|---|
{ "name": "MirrorInfo", "type": "MirrorInfo", "versions": "8+", "nullableVersions": "8+", "ignorable": true,
"about": "Mirror information for creating a mirror topic from a source cluster.", "fields": [
{ "name": "TopicId", "type": "uuid", "versions": "8+",
"about": "The topic ID from the source cluster." }
]} |
The topic ID field ensures mirror topics retain the same topic ID as the source cluster topic. This allows fetch requests to pass validation on the source broker, and enables the system to verify that a topic being mirrored to a same-named topic in the destination cluster is indeed the same logical topic, not a name collision.
In normal topic creation, the MirrorInfo field will be null. When receiving the CreateTopic request, the controller will check the new field. If it is not set, the topic ID will be generated with random UUID as usual. Otherwise, the controller will do the following validation:
- This topic ID is not used by other topics in the current cluster
- The replicas for the partition assignment are all active and not in fenced or controlled shutdown. This is to make sure when a topic gets deleted and re-created with the same topic ID, the stale offline log dir won’t be treated as the active log dir after it becomes online (KAFKA-16234).
Fetch
The Fetch API is extended to add the MirrorLeaderEpoch field used by the destination cluster internal replication.
| Code Block |
|---|
// new request field in FetchPartition type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 2, "ignorable": true,
"about": "The latest known mirror leader epoch." }
// new response field in PartitionData type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 3, "ignorable": true,
"about": "The latest known mirror leader epoch." },
|
CreateMirror
Allows users to create a mirror and supply its configuration. The broker validates that the mirror name is not already in use, contains only permitted characters, and does not end with .stopped or .paused suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "CreateMirrorRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [ |
| Code Block |
{ "apiKey": TBD, "type": "response", "name": "StartMirrorTopicsResponse", // 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": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName", "about": "The cluster mirror name." }, { "name": "TopicsConfig", "type": "[]TopicResultMirrorConfig", "versions": "0+", "about": "The resultscluster formirror the topicsconfigurations.", "fields": [ { "name": "Name", "type": "string", "versions": "0+", "entityTypemapKey": "topicName"true, "about": "The configuration topickey name." }, { "name": "ErrorCodeValue", "type": "int16string", "versions": "0+", "nullableVersions": "0+", "about": "The error code, or 0 if there was no error." value to set for the configuration key."} ]} ] } |
StopMirrorTopics
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 .stopped suffix to the mirror.name topic config to mark the topics as no longer mirrored.
Request
| Code Block |
|---|
{ "apiKey": TBD, "type": "requestresponse", "listeners": ["broker", "controller"], "name": "StopMirrorTopicsRequestCreateMirrorResponse", // Version 0 is the initial version. "validVersions": "0", "flexibleVersions": "0+", "fields": [", "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": "MirrorNameErrorCode", "type": "stringint16", "versions": "0+", "about": "The cluster mirror nameerror code, or 0 if there was no error." }, { "name": "TopicsErrorMessage", "type": "[]TopicDatastring", "versions": "0+", "nullableVersions": "0+", "about": "The data for the topics.", "fields": [ error message, or null if there was no error." } ] } |
StartMirrorTopics
Start mirroring for the specified topics. The broker validates that all target topic partitions are in either UNKNOWN or STOPPED state; otherwise, the request is rejected. Once validated, the request is forwarded to the controller, which sets the mirror.name topic config to the specified mirror name.
| Code Block |
|---|
{ "apiKey": TBD, "type": "request", "listeners": ["broker", "controller"], { "name": "TopicIdStartMirrorTopicsRequest", // Version 0 is the initial version. "typevalidVersions": "uuid0", "versionsflexibleVersions": "0+", "aboutfields": "The unique topic ID."},[ { "name": "TopicNameMirrorName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicNamemirrorName", "about": "The cluster topicmirror name." } ]}, { "name": "PatternsTopics", "type": "[]stringTopicData", "versions": "0+", "about": "PatternsThe todata updatefor inthe mirror.topics.include/exclude.", } ] } |
Response
| Code Block |
|---|
{ "apiKeyfields": TBD,[ "type { "name": "responseTopicId", "nametype": "StopMirrorTopicsResponseuuid", // Version 0 is the initial version. "validVersions"versions": "0+", "flexibleVersionsabout": "0+"The unique topic ID."}, "fields": [ { "name": "ThrottleTimeMsTopicName", "type": "int32string", "versions": "0+", "mapKey": true, "aboutentityType": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quotatopicName", "about": "The topic name." }, { "name": "ErrorCodeNumPartitions", "type": "int16int32", "versions": "0+", "about": "The error code, or 0 if there was no error." number of partitions for the topic. Must match the source topic." } ]}, { "name": "ErrorMessageIncludePatterns", "type": "[]string", "versions": "0+", "nullableVersions": "0+", "default": "null", "about": "TheRegex top-levelpatterns errorto message,add or null if there was no errorto mirror.topics.include." }, { "name": "MirrorNameExcludePatterns", "type": "[]string", "versions": "0+", "entityType": "mirrorName", "about": "The cluster mirror nameRegex patterns to add to mirror.topics.exclude." } ] } { "apiKey": TBD, "type": "response", { "name": "TopicsStartMirrorTopicsResponse", "type": "[]TopicResult", "versions // Version 0 is the initial version. "validVersions": "0", "aboutflexibleVersions": "The results for the topics.0+", "fields": [ { "name": "NameThrottleTimeMs", "type": "stringint32", "versions": "0+", "entityTypeabout": "topicName", "about": "The topic nameThe 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." }, ]} ] } |
PauseMirrorTopics
Pauses data replication and metadata sync for the specified mirror topics, keeping them read-only on the destination cluster.
Request
| Code Block |
|---|
{
"apiKeyname": TBD"ErrorMessage",
"type": "requeststring",
"listenersversions": ["broker0+", "controller"],
"name"nullableVersions": "0+", "default": "PauseMirrorTopicsRequestnull",
// Version 0 is the initial version.
"validVersions"about": "0",
"flexibleVersions": "0+",
"fields": [The top-level error message, or null if there was no error." },
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name to pause the topics for." },
{ "name": "Topics", "type": "[]TopicDataTopicResult", "versions": "0+",
"about": "The dataresults for the topics.",
"fields": [
{ "name": "TopicIdName", "type": "uuidstring", "versions": "0+", "entityType": "topicName",
"about": "The unique topic IDname." },
{ "name": "TopicNameErrorCode", "type": "stringint16", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The topic name error code, or 0 if there was no error." }
]}
]
} |
Response
]}
]
} |
StopMirrorTopics
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 .stopped suffix to the mirror.name topic config to mark the topics as no longer mirrored.
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response"request",
"listeners": ["broker", "controller"],
"name": "PauseMirrorTopicsResponseStopMirrorTopicsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMsMirrorName", "type": "int32string", "versions": "0+",
"about": "The durationcluster inmirror milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
"fields": [
{ "name": "ErrorCodeTopicId", "type": "int16uuid", "versions": "0+",
"about": "The errorunique code, or 0 if there was no errortopic ID." },
{ "name": "ErrorMessageTopicName", "type": "string", "versions": "0+", "nullableVersionsmapKey": "0+"true, "defaultentityType": "nulltopicName",
"about": "The top-level error message, or null if there was no error." topic name." }
]},
{ "name": "MirrorNamePatterns", "type": "[]string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror namePatterns to update in mirror.topics.include/exclude." },
]
}
{
"nameapiKey": "Topics"TBD,
"type": "[]TopicResultresponse",
"versionsname": "0StopMirrorTopicsResponse",
// Version 0 is the initial version.
"aboutvalidVersions": "The0",
results for the topics."flexibleVersions": "0+",
"fields": [
{ "name": "NameThrottleTimeMs", "type": "stringint32", "versions": "0+",
"entityTypeabout": "topicName",
"about": "The topic nameThe 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." }
]}
]
} |
ResumeMirrorTopics
Resumes data replication and metadata sync for previously paused mirror topics from where they left off.
Request
| Code Block |
|---|
{ "apiKey": TBD, no error." }, { "name": "ErrorMessage", "type": "requeststring", "listenersversions": ["broker0+", "controller"], "namenullableVersions": "0+", "default": "ResumeMirrorTopicsRequestnull", // Version 0 is the initial version. "validVersionsabout": "0", "flexibleVersions": "0+", "fields": [The top-level error message, or null if there was no error." }, { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName", "about": "The cluster mirror name." }, { "name": "Topics", "type": "[]TopicDataTopicResult", "versions": "0+", "about": "The dataresults for the topics.", "fields": [ { "name": "TopicIdName", "type": "uuidstring", "versions": "0+", "entityType": "topicName", "about": "The unique topic IDname." }, { "name": "TopicNameErrorCode", "type": "stringint16", "versions": "0+", "mapKey": true, "entityType "about": "topicName", The error "about": "The topic namecode, or 0 if there was no error." } ]} ] } |
...
PauseMirrorTopics
Pauses data replication and metadata sync for the specified mirror topics, keeping them read-only on the destination cluster.
| Code Block |
|---|
{
"apiKey": 103TBD,
"type": "request",
"listeners": "response"["broker", "controller"],
"name": "ResumeMirrorTopicsResponsePauseMirrorTopicsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMsMirrorName", "type": "int32string", "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." },
mirror name to pause the topics for." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
"fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
{ "name": "ErrorCodeTopicName", "type": "int16string", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The error code, or 0 if there was no error." },
{ "name": "ErrorMessage", topic name." }
]}
]
}
{
"apiKey": TBD,
"type": "stringresponse",
"versionsname": "0+PauseMirrorTopicsResponse",
"nullableVersions // Version 0 is the initial version.
"validVersions": "0+",
"defaultflexibleVersions": "null0+",
"aboutfields": "The top-level error message, or null if there was no error." },[
{ "name": "MirrorNameThrottleTimeMs", "type": "stringint32", "versions": "0+",
"entityTypeabout": "mirrorName",
"about": "The cluster mirror nameThe 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": "TopicsErrorCode", "type": "[]TopicResultint16", "versions": "0+",
"about": "The results for the topics.", "fields": [
error code, or 0 if there was no error." },
{ "name": "NameErrorMessage", "type": "string", "versions": "0+", "entityTypenullableVersions": "topicName0+",
"default": "null",
"about": "The topic name top-level error message, or null if there was no error." },
{ "name": "ErrorCodeMirrorName", "type": "int16string", "versions": "0+",
"entityType": "mirrorName",
"about": "The error code, or 0 if there was no error." }
]}
]
} |
DeleteMirror
...
cluster |
...
Request
| Code Block |
|---|
{ "apiKey": 104, "type mirror name." }, { "name": "requestTopics", "listenerstype": "["broker",]TopicResult", "versions": "controller0"], "nameabout": "DeleteMirrorRequest",The results for the topics.", "fields": [ // Version 0 is the initial version. "validVersions{ "name": "Name", "type": "string", "versions": "0", "flexibleVersionsentityType": "0+topicName", "fieldsabout": [ "The topic name." }, { "name": "MirrorNameErrorCode", "type": "stringint16", "versions": "0+", "entityType": "mirrorName", "about": "The cluster mirror name to delete."error code, or 0 if there was no error." } ]} } |
...
]
} |
ResumeMirrorTopics
Resumes data replication and metadata sync for previously paused mirror topics from where they left off.
| Code Block |
|---|
{
"apiKey": 104TBD,
"type": "response"request",
"listeners": ["broker", "controller"],
"name": "DeleteMirrorResponseResumeMirrorTopicsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMsMirrorName", "type": "int32string", "versions": "0+",
"about": "The durationcluster in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quotamirror name." },
{ "name": "ErrorCodeTopics", "type": "int16[]TopicData", "versions": "0+",
"about": "The errordata code,for or 0 if there was no errorthe topics." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
"about": "The error message, or null if there was no error." }
]
} |
ListMirrors
Returns the current mirror names and their associated topic counts in the cluster. It also includes source cluster ID and bootstrap server.
Request
| Code Block |
|---|
{ "apiKey": TBD, "fields": [ { "name": "TopicId", "type": "requestuuid", "listenersversions": ["broker0+"], "nameabout": "ListMirrorsRequest"The unique topic ID."}, // Version 0 is the initial version. "validVersions{ "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "flexibleVersionsentityType": "0+topicName", "fieldsabout": [ "The topic name." } ]} ] } |
Response
| Code Block |
|---|
{ "apiKey": TBD, "type": "response", "name": "ListMirrorsResponseResumeMirrorTopicsResponse", // 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": "[]ListedMirror", "versions": "0+", "about": "Each mirror in the response.", "fields": [ if there was no error." }, { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName", "about": "The cluster mirror name." }, { "name": "SourceBootstrapTopics", "type": "string[]TopicResult", "versions": "0+", "about": "The sourceresults clusterfor bootstrapthe serverstopics.", "fields": },[ { "name": "SourceClusterIdName", "type": "string", "versions": "0+", "defaultentityType": "topicName", "about": "The source cluster ID, or empty if not yet resolvedtopic name." }, { "name": "TopicCountErrorCode", "type": "int32int16", "versions": "0+", "default": "0", "about": "The numbererror of topics configured for this mirror. 0 indicates an empty mirror with no topicscode, or 0 if there was no error." } ]} ] } |
DescribeMirrors
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.
...
DeleteMirror
Permanently deletes a cluster mirror, including its configuration. The mirror must be empty (no topics) or all its partitions must be in STOPPED state. After deletion, all metadata are tombstoned, making failback impossible. This is an irreversible operation.
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "brokercontroller"],
"name": "DescribeMirrorsRequestDeleteMirrorRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorNamesMirrorName", "type": "[]string", "versions": "0+", "entityType": "mirrorName",
"about": "The namescluster ofmirror the mirrorsname to describe. Null or empty array means all mirrors." },
{ "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+", "default": "false",
"about": "Whether to include authorized operations." }
]
} |
Response
| Code Block |
|---|
delete."} ] } { "apiKey": TBD, "type": "response", "name": "DescribeMirrorsResponseDeleteMirrorResponse", // 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", "type": "Mirrorsstring", "typeversions": "[]DescribedMirror0+", "versionsnullableVersions": "0+", "about": "EachThe describederror mirror."message, "fields": [ { "name": "ErrorCode", or null if there was no error." } ] } |
ListMirrors
Returns the current mirror names and their associated topic counts in the cluster. It also includes source cluster ID and bootstrap server.
| Code Block |
|---|
{ "apiKey": TBD, "type": "int16request", "versionslisteners": ["0+broker"], "aboutname": "TheListMirrorsRequest", error code,// orVersion 0 is ifthe thereinitial wasversion. no error."validVersions": }"0", { "nameflexibleVersions": "MirrorName0+", "typefields": "string", "versions[] } { "apiKey": TBD, "type": "0+response", "entityTypename": "mirrorNameListMirrorsResponse", // Version 0 is the initial version. "aboutvalidVersions": "0"The, cluster mirror name." }"flexibleVersions": "0+", "fields": [ { "name": "AuthorizedOperationsThrottleTimeMs", "type": "int32", "versions": "0+", "defaultabout": "-2147483648", The duration in milliseconds for which the request "about": "32-bit bitfield to represent authorized operations for this mirrorwas throttled due to a quota violation, or zero if the request did not violate any quota." }, { "name": "TopicsErrorCode", "type": "[]TopicPartitionsint16", "versions": "0+", "about": "EachThe topicerror incode, the mirror.", "fields": [ or 0 if there was no error." }, { "name": "TopicNameErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null", "about": "The topic nametop-level error message, or null if there was no error." }, { "name": "PartitionsMirrors", "type": "[]PartitionDetailListedMirror", "versions": "0+", "about": "Each mirror in partitionthe detailresponse.", "fields": [ { "name": "PartitionIndexMirrorName", "type": "int32string", "versions": "0+", "entityType": "0+mirrorName", "about": "The partitioncluster mirror indexname." }, { "name": "SourceOffsetSourceBootstrap", "type": "int64string", "versions": "0+", "default": "-1", "about": "The high watermark offset from the source cluster leader, or -1 if not yet availablebootstrap servers." }, { "name": "DestinationOffsetSourceClusterId", "type": "int64string", "versions": "0+", "default": "-1", "about": "The logsource end offset on the destination clustercluster ID, or -1empty if not yet availableresolved." }, { "name": "LagTopicCount", "type": "int64int32", "versions": "0+", "default": "-10", "about": "The number of topics configured for this mirror. 0 indicates an empty mirror with no topics." } ]} ] } |
DescribeMirrors
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.
| Code Block |
|---|
{ "apiKey": TBD, "type": "request", "listeners": ["broker"], "name": "DescribeMirrorsRequest", // Version 0 is the initial version. "validVersions": "0", "flexibleVersions": "0+", "fields": [ lag (source offset - destination offset), or -1 if not yet available." }, { "name": "State", "type": "string", "versions": "0+", "about": "The partition state." }, { "name": "MirrorNames", "type": "int32[]string", "versions": "0+", "defaultentityType": "-1mirrorName", "about": "The lastnames mirrorof leaderthe epoch,mirrors orto -1describe. ifNull notor available."empty }array means ]}all mirrors." }, { ]} ]} ] } |
BumpLeaderEpochs
Sets a minimum leader epoch on the specified partitions. The controller increments each partition's leader epoch to at least the requested value.
Request
| Code Block |
|---|
{ "apiKey": TBD, "type": "request", "listeners": ["broker", "controller"]"name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+", "default": "false", "about": "Whether to include authorized operations." } ] } { "apiKey": TBD, "type": "response", "name": "BumpLeaderEpochsRequestDescribeMirrorsResponse", // Version 0 is the initial version. "validVersions": "0", "flexibleVersions": "0+", "fields": [ { "name": "TopicsThrottleTimeMs", "type": "[]TopicStateint32", "versions": "0+", "about": "The topic and partitions state."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": "fields": ["ErrorCode", "type": "int16", "versions": "0+", "about": "The error code, or 0 if there was no error." }, { {"name"name": "ErrorMessage", "type": "string", "versions": "TopicId0+", "typenullableVersions": "uuid0+", "versionsdefault": "0+null", "about": "The uniquetop-level topic ID."}, error message, or null if there was no error." }, { "name": "PartitionsMirrors", "type": "[]LeaderEpochStateDescribedMirror", "versions": "0+", "about": "TheEach partitiondescribed leader epochsmirror.", "fields": [ { {"name": "partitionIndexErrorCode", "type": "int32int16", "versions": "0+", "about": "The partition index."}, error code, or 0 if there was no error." }, { "name": "minLeaderEpochMirrorName", "type": "int32string", "versions": "0+", "defaultentityType": -1"mirrorName", "about": "The minimum leader epoch that the destination cluster shouldmirror bump toname."} ]}, ]} ] } |
Response
| Code Block |
|---|
{ "apiKey": TBD, "type{ "name": "AuthorizedOperations", "type": "int32", "versions": "response0+", "namedefault": "BumpLeaderEpochsResponse-2147483648", // Version 0 is the initial version. "validVersionsabout": "0", "flexibleVersions": "0+", "fields": [ 32-bit bitfield to represent authorized operations for this mirror." }, { "name": "ThrottleTimeMsTopics", "type": "int32[]TopicPartitions", "versions": "0+", "about": "TheEach durationtopic in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }, the mirror.", "fields": [ { "name": "ErrorCodeTopicName", "type": "int16string", "versions": "0+", "about": "The error code, or 0 if there was no errortopic name." }, { "name": "TopicsPartitions", "type": "[]TopicPartitionsPartitionDetail", "versions": "0+", "about": "Each topic in the mirrorpartition detail.", "fields": [ { "name": "TopicNamePartitionIndex", "type": "stringint32", "versions": "0+", "about": "The topicpartition nameindex." }, { "name": "SourceOffset", "type": "Partitionsint64", "typeversions": "[]PartitionDetail0+", "versionsdefault": "0+-1", "about": "Each partition state.", "fields": [ The high watermark offset from the source cluster leader, or -1 if not yet available." }, { "name": "PartitionIndexDestinationOffset", "type": "int32int64", "versions": "0+", "default": "-1", "about": "The partition index log end offset on the destination cluster, or -1 if not yet available." }, { "name": "ErrorCodeLag", "type": "int16int64", "versions": "0+", "default": "-1", "about": "The errorlag code,(source oroffset 0- ifdestination thereoffset), wasor no error." } ]} ]} ] } |
ReadMirrorStates
Reads the current mirror partition states from the internal __mirror_state topic on the destination cluster.
Request
| Code Block |
|---|
{ "apiKey": TBD, "type": "request", "listeners": ["broker", "controller"], "name": "ReadMirrorStatesRequest", // Version 0 is the initial version. "validVersions": "0", "flexibleVersions": "0+", "fields": [ -1 if not yet available." }, { "name": "State", "type": "string", "versions": "0+", "about": "The partition state." }, { "name": "MirrorName", "type": "stringint32", "versions": "0+", "entityType"default": "-1", "about": "mirrorName", "about": "The cluster mirror name." }, { "name": "Topics", "type": "[]TopicData", "versions": "0", "about": "The data for the topics.", "fields": [ { "name": "Name", "type": "string", "versionsThe last mirror leader epoch, or -1 if not available." } ]} ]} ]} ] } |
BumpLeaderEpochs
Sets a minimum leader epoch on the specified partitions. The controller increments each partition's leader epoch to at least the requested value.
| Code Block |
|---|
{ "apiKey": TBD, "type": "request", "listeners": ["broker", "controller"], "name": "BumpLeaderEpochsRequest", // Version 0 is the initial version. "validVersions": "0", "entityTypeflexibleVersions": "topicName0+", "about"fields": "The topic name." },[ { "name": "PartitionsTopics", "type": "[]PartitionDataTopicState", "versions": "0+", "about": "The datatopic forand thepartitions partitionsstate.", "fields": [ { "name": "PartitionIndexTopicId", "type": "int32uuid", "versions": "0+", "about": "The unique partitiontopic indexID." }, ]} ]} ] } |
Response
| Code Block |
|---|
{ "apiKey": TBD, { "name": "Partitions", "type": "response[]LeaderEpochState", "versions": "name"0+", "about": "ReadMirrorStatesResponseThe partition leader epochs.", // Version 0 is the initial version. "validVersionsfields": "0",[ "flexibleVersions": "0+", "fields": [ { "name": "ThrottleTimeMspartitionIndex", "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." }, { partition index."}, {"name": "ErrorCodeminLeaderEpoch", "type": "int16int32", "versions": "0+", "default": -1, "about": "The minimum errorleader code,epoch orthat 0the ifdestination therecluster wasshould nobump errorto."} ]}, ]} ] } { "nameapiKey": "ErrorMessage"TBD, "type": "stringresponse", "versionsname": "0+BumpLeaderEpochsResponse", // Version "nullableVersions0 is the initial version. "validVersions": "0+", "defaultflexibleVersions": "null0+", "aboutfields": "The top-level error message, or null if there was no error." }, [ { "name": "TopicsThrottleTimeMs", "type": "[]TopicResultint32", "versions": "0+", "about": "The duration readin resultsmilliseconds for which the topics.", "fields": [ request was throttled due to a quota violation, or zero if the request did not violate any quota." }, { "name": "NameErrorCode", "type": "stringint16", "versions": "0+", "entityType": "topicName", "about": "The topic nameerror code, or 0 if there was no error." }, { "name": "PartitionsTopics", "type": "[]PartitionResultTopicPartitions", "versions": "0+", "about": "TheEach resultstopic forin the partitionsmirror.", "fields": [ { "name": "PartitionIndexTopicName", "type": "int32string", "versions": "0+", "about": "The partitiontopic indexname." }, { "name": "LastMirrorEpochPartitions", "type": "int32[]PartitionDetail", "versions": "0", default": "-1"+", "about": "TheEach last mirror leader epoch, or -1 if not available." }, partition state.", "fields": [ { "name": "StatePartitionIndex", "type": "int8int32", "versions": "0+", "about": "The mirror partition stateindex." }, { "name": "ErrorCode", "type": "int16", "versions": "0", "about": "The error code, or 0 if there was no error." } ]} ]} ] } |
...
ReadMirrorStates
Persists Reads the current mirror partition state transitions to states from the internal __mirror_state topic on the destination cluster.
Request
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "WriteMirrorStatesRequestReadMirrorStatesRequest",
// 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.
"validVersions": "0",
"aboutflexibleVersions": "The data for the topics.0+",
"fields": [
{ "name": "NameMirrorName", "type": "string", "versions": "0+", "entityType": "topicNamemirrorName",
"about": "The cluster topicmirror name." },
{ "name": "PartitionsTopics", "type": "[]PartitionDataTopicData", "versions": "0",
"about": "The data for the partitionstopics.", "fields": [
{ "name": "PartitionIndexName", "type": "int32string", "versions": "0",
"entityType": "topicName",
"about": "The partitiontopic indexname." },
{ "name": "Partitions", "type": "int32[]PartitionData", "versions": "0", "default": "-1",
"about": "The lastdata mirrorfor leader epoch, or -1 if not available." },
{ "name": "State", "type": "int8", "versions": "0+",
"about": "The mirror partition state." }the partitions.", "fields": [
]}
]},
{ "name": "StoppedTopicsPartitionIndex", "type": "[]stringint32", "versions": "0+",
"about": "The topic names to be stopped." }
partition index." }
]}
]}
]
} |
Response
| Code Block |
|---|
{ "apiKey": TBD, "type": "response", "name": "WriteMirrorStatesResponseReadMirrorStatesResponse", // 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 writeread 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", "fields": [ : "PartitionIndex", "type": "int32", "versions": "0", "about": "The partition index." }, { "name": "LastMirrorEpoch", "type": "int32", "versions": "0", default": "-1", "about": "The last mirror leader epoch, or -1 if not available." }, { "name": "PartitionIndexState", "type": "int32int8", "versions": "0+", "about": "The mirror partition indexstate." }, { "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 controller so that it is at least as high as the source cluster's leader epoch.
Request
WriteMirrorStates
Persists mirror partition state transitions to the internal __mirror_state topic on the destination cluster.
| 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." },
|
| Code Block |
{ "apiKey": TBD, "type": "request", "listeners": ["broker", "controller"], "name": "BumpLeaderEpochsRequest", // Version 0 is the initial version. "validVersions": "0", "flexibleVersions": "0+", "fields": [ { "name": "TopicsPartitions", "type": "[]TopicStatePartitionData", "versions": "0+", "about": "The topicdata for andthe partitions state.", "fields": [ { "name": "TopicIdPartitionIndex", "type": "uuidint32", "versions": "0+", "about": "The uniquepartition topic IDindex." }, { "name": "Partitions", "type": "[]LeaderEpochStateint32", "versions": "0+", "aboutdefault": "The partition leader epochs.-1", "fieldsabout": [ "The last mirror leader epoch, or -1 if not available." }, { "name": "partitionIndexState", "type": "int32int8", "versions": "0+", "about": "The mirror partition indexstate." }, ]} ]}, { "name": "minLeaderEpochStoppedTopics", "type": "int32[]string", "versions": "0+", "default": -1, "about": "The minimumtopic leadernames epochto that the destination cluster should bump tobe stopped." } ]} ]} ] } |
Response
| Code Block |
|---|
] } { "apiKey": TBD, "type": "response", "name": "BumpLeaderEpochsResponseWriteMirrorStatesResponse", // 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 quotanot violate any quota." }, { "name": "ErrorCode", "type": "int16", "versions": "0+", "about": "The error code, or 0 if there was no error." }, { "name": "ErrorCodeErrorMessage", "type": "int16string", "versions": "0+", "nullableVersions": "0+", "default": "null", "about": "The top-level error codemessage, or 0null if there was no error." }, { "name": "Topics", "type": "[]TopicPartitionsTopicResult", "versions": "0+", "about": "EachThe write topicresults infor the mirrortopics.", "fields": [ { "name": "TopicNameName", "type": "string", "versions": "0+", "entityType": "topicName", "about": "The topic name." }, { "name": "Partitions", "type": "[]PartitionDetailPartitionResult", "versions": "0+", "about": "Each partition stateThe 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." } ]} ]} ] } |
Cluster Metadata Records
This section describes records written to the KRaft metadata log by the active controller as part of Cluster Mirroring operations.
PartitionChangeRecord
The new version adds MinLeaderEpoch to support the Leader Epoch Bump (LEB) during failoverWritten by the controller when processing a BumpLeaderEpochs request. The record carries a minLeaderEpoch field that sets a floor for the partition's leader epoch.
| Code Block |
|---|
{
"apiKey": 5,
"type": "metadata",
"name": "PartitionChangeRecord",
"validVersions": "0-3",
"flexibleVersions": "0+",
"fields": [
// ... existing fields ...
{"name": "MinLeaderEpoch", "type": "int32", "versions": "3+", "default": -1,
"about": "The minimum leader epoch requested."}
]
} |
MirrorPidResetRecord
A control record (type MIRROR_PID_RESET) written to each partition's data log during the STOPPING transition.
| Code Block |
|---|
{ "type": "data", "name": "MirrorPidResetRecord", "validVersions": "0", "flexibleVersions": "0+", "fields": [ // ... existing fields ...{ "name": "Version", "type": "int16", "versions": "0", { "nameabout": "MinLeaderEpoch", "typeThe version of the mirror PID reset record."}, { "name": "int32SourceClusterId", "versionstype": "3+string", "defaultversions": -1"0", "about": "The minimumsource cluster leaderUUID epochfor requestedverification."} ] } |
Mirror Metadata Records
LastMirrorEpochs
This section describes records written to the __mirror_state internal topic by the MirrorCoordinator to track mirror state and synchronization points across brokers.
LastMirrorEpochs
Written during the STOPPING transition to record the last mirrored leader epoch for each partition before the destination becomes writableThe greatest leader epoch of a given partition that a destination cluster recognizes from the source cluster.
| Code Block |
|---|
{
"apiKey": 1,
"type": "coordinator-key",
"name": "LastMirrorEpochsKey",
"validVersions": "0",
"flexibleVersions": "none",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0",
"about": "The cluster mirror name."}
]
}
{
"apiKey": 1,
"type": "coordinator-value",
"name": "LastMirrorEpochsValue",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Topics", "type": "[]Topic", "versions": "0+",
"about": "The mirror topics for which we want to store the last mirror 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 mirror epochs.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "", "type": "int32", "versions": "0+",
"about": "The last mirror leader epoch for this partition." }
]}
]}
]
} |
MirrorPartitionState
MirrorPartitionState record represents the lifecycle states of a Written on every state transition. Tracks the current state of each mirrored partition.
| Code Block |
|---|
{
"apiKey": 2,
"type": "coordinator-key",
"name": "MirrorPartitionStateKey",
"validVersions": "0",
"flexibleVersions": "none",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0",
"about": "The cluster mirror name."}
]
}
{
"apiKey": 2,
"type": "coordinator-value",
"name": "MirrorPartitionStateValue",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "TopicName", "type": "string", "versions": "0",
"about": "The topic name."},
{ "name": "Partition", "type": "int32", "versions": "0",
"about": "The partition index."},
{ "name": "State", "type": "int8", "versions": "0+",
"about": "The mirror partition state." }
]
} |
Type Enumerations
EntityType
Enumerations
This section lists the new values added to existing Kafka type enumerations to support Cluster Mirroring
Configuration
Represents a mirror as a configurable resource in cluster metadata. Mirror-level properties such as source cluster bootstrap servers, security credentials, mirror.topics.include, and mirror.topics.exclude are stored under this type, keyed by mirror name.
| Code Block | ||
|---|---|---|
| ||
public enum Type {
// existing types unchanged
MIRROR((byte) 64, "mirror");
} |
Entity
A schema-level annotation for MirrorName string fields in protocol messages. The message generator uses it to validate that mirror name fields across all request/response schemas conform to the expected type.A new entity type is added for the message generator to provide schema-level type validation for mirror name fields:
| Code Block | ||
|---|---|---|
| ||
public enum EntityType {
// existing types unchanged
@JsonProperty("mirrorName")
MIRROR_NAME(FieldType.StringFieldType.INSTANCE);
} |
...
Resource
A new resource type is added to the ResourceType enum to enable per-mirror authorization:An ACL resource type that represents a cluster mirror as a securable object. Authorization checks use this type with the mirror name as the resource name.
| Code Block | ||
|---|---|---|
| ||
public enum ResourceType {
// existing types unchanged
CLUSTER_MIRROR((byte) 8); |
CoordinatorType
Coordinator
A coordinator type for locating the broker responsible for a given mirror name. The coordinator partition is determined by hashing the mirror name across __mirror_state topic partitions.The FindCoordinatorRequest object is extended to support a new coordinator type:
| Code Block | ||
|---|---|---|
| ||
public enum CoordinatorType {
// existing types unchanged
MIRROR((byte) 3);
} |
...
A new configuration resource type is added for cluster mirrors, which is stored in the cluster metadata internal log.
| Code Block | ||
|---|---|---|
| ||
public enum Type {
// existing types unchanged
MIRROR((byte) 64, "mirror");
} |
Configuration
This section describes all new configurations introduced by Cluster Mirroring.
Mirror Configuration
Set via CreateMirror or IncrementalAlterConfigs. Stored in cluster metadata records.
...