...
Create a new cluster mirror configuration in the destination cluster (forbidden suffixes: .removed, .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 |
Add a topic or set of topics to an existing cluster mirror (start mirroring; the topic flag accepts regex expression to operate on multiple topics at once):
| Code Block |
|---|
|
$ bin/kafka-mirror.sh --bootstrap-server :9094 --add --topic my-topic --mirror my-mirror --replication-factor 2
Added 1 topic(s) to mirror my-mirror: [my-topic] |
List configured mirrors with additional informationRemove a specific topic or set of topics from a mirror (failover; topics become writable):
| Code Block |
|---|
|
$ bin/kafka-mirrorsmirror.sh --bootstrap-server :9094 --remove --topic my-topic --mirror my-mirror
Removed 1 topic(s) from mirror my-mirror: [my-topic] |
Delete a mirror including its topics and configuration (the mirror must be empty or include only stopped partitions):
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
MIRROR TOPICS list
MIRROR TOPICS CLUSTER-ID BOOTSTRAP-SERVER
my-mirror 2 lBq12jYZRp-9wF3M9MPopg localhost:9091,localhost:9092
new-mirror 1 lBq12jYZRp-9wF3M9MPopg localhost:9091,localhost:9092 |
Describe configured mirrors to check their lag compared to their source topics (add use --mirror flag to filter out other partitions from other mirrors):
| Code Block |
|---|
|
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --describe
MIRROR TOPIC PARTITION SOURCE-OFFSET DESTINATION-OFFSET LAG STATE
my-mirror bar 0 2324 2324 0 MIRRORING
my-mirror foo 0 69 66 3 MIRRORING
my-mirror foo 1 94 84 10 MIRRORING
my-mirror foo 2 94 90 4 MIRRORING
new-mirror baz 0 189 189 0 MIRRORING
new-mirror baz 1 859 859 0 MIRRORING |
Pause mirroring for a specific topic or set of topics (topics remain read-onlyAlter mirror configuration (any valid configuration triggers a reconnection):
| Code Block |
|---|
|
$ bin/kafka-mirrorsconfigs.sh --bootstrap-server :9094 --entity-pausetype mirrors --topicentity-name my-topicmirror \
--mirroralter my-mirror
Paused mirroring for 1 topic(s): [my-topic]--add-config bootstrap.servers=localhost:9092
Completed updating config for mirror my-mirror. |
Throttling on the destination clusterResume mirroring for a specific topic or set of topics:
| Code Block |
|---|
|
$ bin/kafka-mirrorsconfigs.sh --bootstrap-server :9094 --resumeentity-type brokers ---topic my-topic --mirror my-mirror
Resumed mirroring for 1 topic(s): [my-topic] |
Remove a specific topic or set of topics from a mirror (failover; topics become writable):
| Code Block |
|---|
|
entity-name 4 \
--alter --add-config mirror.replication.throttled.rate=100000000
Completed updating config for broker 4.
$ bin/kafka-mirrorconfigs.sh --bootstrap-server :9094 --entity-removetype topics --topic my-topic --mirrorentity-name my-mirror
Removed 1 topic(s) from mirror my-mirror: [my-topic] |
Delete a mirror including its topics and configuration (the mirror must be empty or include only stopped partitions):
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:9092mirror.replication.throttled.replicas=[0:4]
Completed updating config for mirrortopic my-mirrortopic. |
Throttling on the destination source cluster:
| Code Block |
|---|
|
$ bin/kafka-configs.sh --bootstrap-server :90949091 --entity-type brokersalter --entityadd-name 4config 'consumer_byte_rate=1024' \
--alterentity-type clients --addentity-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-name broker-4-fetcher-0-mirror-my-mirror
Completed updating config for client broker-4-fetcher-0-mirror-my-mirror. |
...
| 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);
/**
* Add topics to an existing cluster mirror for cross-cluster replication.
*
* When topics are added to 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 topicToMirrorNamemirrorName MapThe ofmirror topic namesname to mirroradd names, allowing multiple the topics to be
* @param topics Set of topic names added to potentially different mirrors in a single operation
to add to mirroring
* @param options Options for the add topics to mirror operation
* @return The AddTopicsToMirrorResult containing futures for each topic addition
*/
AddTopicsToMirrorResult addTopicsToMirror(Map<StringString mirrorName, String>Set<String> topicToMirrorNametopics, AddTopicsToMirrorOptions options);
/**
* Remove topics from cluster mirror, making them writable on the destination cluster.
*
* 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). Removing topics from
* the mirror clears the mirrorName field from partition metadata, which allows producers to write
* to these partitions.
*
* @param mirrorName The mirror name to remove the topics from
* @param topics Set of topic names to remove from mirroring
* @param options Options for the remove topics from mirror operation
* @return The RemoveTopicsFromMirrorResult containing futures for each topic removal
*/
RemoveTopicsFromMirrorResult removeTopicsFromMirror(String mirrorName, Set<String> topics, RemoveTopicsFromMirrorOptions options);
/**
* 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 mirror name to pause the topics for
* @param topics Set of topic names to pause mirroring for
* @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 mirror name to resume the topics for
* @param topics Set of topic names to resume mirroring for
* @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);
/**
* 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);
|
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "CreateMirrorRequest",
"latestVersionUnstable": true,
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
"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."}
]}
]
} |
CreateMirrorResponse
| 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." }
]
} |
AddTopicsToMirror
Adds topics to a specified mirror. The broker validates that all target topic partitions are in either UNKNOWN or STOPPED state; otherwise, the request is rejected with an INVALID_REQUEST error. Once validated, the request is forwarded to the controller, which sets the mirror.name topic config to the specified mirror name.
AddTopicsToMirrorRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "AddTopicsToMirrorRequest",
// 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." }
]}
]
} |
AddTopicsToMirrorResponse
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "AddTopicsToMirrorResponse",
// 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": "ConfigTopics", "type": "[]MirrorConfigTopicResult", "versions": "0+",
"about": "The clusterresults for mirrorthe configurationstopics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "mapKeyentityType": true"topicName",
"about": "The configuration keytopic name." },
{ "name": "ValueErrorCode", "type": "stringint16", "versions": "0+",
"nullableVersionsabout": "0+",The error code, or 0 if there was no error." }
"about": "The value to set for the configuration key."}
]}
]
} |
...
RemoveTopicsFromMirror
Allows users to detach topics from their associated mirror. The broker validates that all target topic partitions are in either PREPARING or MIRRORING state. Once validated, the request is forwarded to the controller, which appends the ".removed" suffix to the mirror.name topic config to mark the topics as no longer mirrored.
RemoveTopicsFromMirrorRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"response" "listeners": ["broker", "controller"],
"name": "CreateMirrorResponseRemoveTopicsFromMirrorRequest",
// 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+", "nullableVersions"mapKey": true, "entityType": "0+topicName",
"about": "The error message, or null if there was no error." topic name." }
]}
]
} |
AddTopicsToMirror
Adds topics to a specified mirror. The broker validates that all target topic partitions are in either UNKNOWN or STOPPED state; otherwise, the request is rejected with an INVALID_REQUEST error. Once validated, the request is forwarded to the controller, which sets the mirror.name topic config to the specified mirror name.
AddTopicsToMirrorRequest
RemoveTopicsFromMirrorResponse
| Code Block |
|---|
{
"apiKey": TBD,
"type": "requestresponse",
"listeners": ["broker", "controller"],
"name": "AddTopicsToMirrorRequestRemoveTopicsFromMirrorResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "TopicsThrottleTimeMs", "type": "[]TopicDataint32", "versions": "0+",
"about": "The duration in datamilliseconds for which the topics.",
"fields": [
request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "TopicIdErrorCode", "type": "uuidint16", "versions": "0+",
"about": "The unique topic ID." error code, or 0 if there was no error." },
{ "name": "TopicNameErrorMessage", "type": "string", "versions": "0+", "mapKeynullableVersions": true"0+", "entityTypedefault": "topicNamenull",
"about": "The topic name top-level error message, or null if there was no error." },
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name."}
]},
]
} |
AddTopicsToMirrorResponse
| Code Block |
|---|
{
"apiKey { "name": TBD"Topics",
"type": "response[]TopicResult",
"nameversions": "AddTopicsToMirrorResponse0",
// Version 0 is the initial version.
"validVersionsabout": "0",
The "flexibleVersions": "0+",
results for the topics.", "fields": [
{ "name": "ThrottleTimeMsName", "type": "int32string", "versions": "0+",
"aboutentityType": "topicName"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"about": "The topic name." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 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]}
]
} |
PauseMirrorTopics
Pauses data replication and metadata sync for the specified mirror topics, keeping them read-only on the destination cluster.
PauseMirrorTopicsRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "PauseMirrorTopicsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
." },
{ "name": "ErrorCodeMirrorName", "type": "int16string", "versions": "0+",
"about": "The errormirror code,name orto 0pause ifthe there was no errortopics for." },
]}
]
} |
RemoveTopicsFromMirror
Allows users to detach topics from their associated mirror. The broker validates that all target topic partitions are in either PREPARING or MIRRORING state. Once validated, the request is forwarded to the controller, which appends the ".removed" suffix to the mirror.name topic config to mark the topics as no longer mirrored.
RemoveTopicsFromMirrorRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "RemoveTopicsFromMirrorRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
"fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+",
"fieldsabout": [
"The unique topic ID."},
{ "name": "TopicsTopicName", "type": "[]TopicDatastring", "versions": "0+", "mapKey": true, "aboutentityType": "topicName"The,
data for the topics.",
"fieldsabout": [ "The topic name." }
]}
]
} |
PauseMirrorTopicsResponse
| Code Block |
|---|
{
"apiKey": TBD,
"nametype": "TopicIdresponse",
"typename": "uuidPauseMirrorTopicsResponse",
// Version 0 is the initial version.
"versionsvalidVersions": "0+",
"aboutflexibleVersions": "The unique topic ID."}0+",
"fields": [
{ "name": "TopicNameThrottleTimeMs", "type": "stringint32", "versions": "0+", "mapKey": true,
"entityTypeabout": "topicName",
"about": "The topic name." }
]}
]
} |
RemoveTopicsFromMirrorResponse
| Code Block |
|---|
{
"apiKey": TBD,
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": "responseint16",
"nameversions": "RemoveTopicsFromMirrorResponse0+",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions"about": "The error code, or 0 if there was no error." },
{ "name": "0+ErrorMessage",
"fieldstype": [
{"string", "nameversions": "ThrottleTimeMs0+", "typenullableVersions": "int320+", "versionsdefault": "0+null",
"about": "The durationtop-level in milliseconds for which the request was throttled due to a quota violationerror message, or zeronull if thethere requestwas did not violate any quotano error." },
{ "name": "ErrorCodeMirrorName", "type": "int16string", "versions": "0+", "entityType": "mirrorName",
"about": "The errorcluster code, or 0 if there was no errormirror name." },
{ "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": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]
} |
LastMirroredOffset
Allows destination cluster partition leaders in PREPARING state to query the LMO from the source cluster. If the source cluster has no record of this offset in its internal topic, it returns 0, meaning the log must be truncated to the beginning and mirroring starts from scratch. This is particularly important during failback. The last mirrored offset identifies where mirrored data ends and un-mirrored data begins. Records beyond this offset must be truncated before mirroring new data from the new source cluster; otherwise, the two clusters would contain inconsistent data.
...
there was no error." }
]}
]
} |
ResumeMirrorTopics
Resumes data replication and metadata sync for previously paused mirror topics from where they left off.
ResumeMirrorTopicsRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "LastMirroredOffsetsRequestResumeMirrorTopicsRequest",
// 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."fields": },[
{ "name": "PartitionsTopicId", "type": "[]PartitionDatauuid", "versions": "0+",
"about": "The dataunique fortopic the partitionsID."}, "fields": [
{ "name": "PartitionIndexTopicName", "type": "int32string", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The partitiontopic indexname." }
]}
]}
]
} |
...
ResumeMirrorTopicsResponse
| Code Block |
|---|
{
"apiKey": TBD103,
"type": "response",
"name": "LastMirroredOffsetsResponseResumeMirrorTopicsResponse",
// 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": "TopicsErrorMessage", "type": "[]TopicResultstring", "versions": "0+",
"aboutnullableVersions": "The results for the topics.0+", "fieldsdefault": ["null",
{ "nameabout": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic nameThe top-level error message, or null if there was no error." },
{ "name": "PartitionsMirrorName", "type": "[]PartitionResultstring", "versions": "0+",
"entityType": "mirrorName",
"about": "The resultscluster formirror the partitionsname.", "fields": [
},
{ "name": "PartitionIndexTopics", "type": "int32[]TopicResult", "versions": "0",
"about": "The results for partitionthe indextopics." },
"fields": [
{ "name": "LastMirroredOffsetName", "type": "int64string", "versions": "0",
"entityType": "topicName",
"about": "The lasttopic mirrored offsetname." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
ListMirrors
Returns the current mirror names and their associated topic counts in the cluster.
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "ListMirrorsResponse",
// 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 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": "ErrorCodeErrorMessage", "type": "int16string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error codemessage, or 0null if there was no error." },
{ "name": "Mirrors", "type": "[]ListedMirror", "versions": "0+",
"about": "Each mirror in the response.", "fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name." },
{ "name": "TopicCount", "type": "int32", "versions": "0+", "default": "0",
"about": "The number of topics configured for this mirror. 0 indicates an empty mirror with no topics." },
{ "name": "ClusterId", "type": "string", "versions": "0+",
"about": "The source cluster ID." },
{ "name": "BootstrapServer", "type": "string", "versions": "0+",
"about": "The source cluster bootstrap servers." }
]}
]
} |
...
| 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 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 (INITIALIZING, PREPARING, MIRRORING, STOPPING, STOPPED, FAILED)." }
]}
]},
{ "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
"about": "32-bit bitfield to represent authorized operations for this mirror." }
]}
]
} |
ReadMirrorStates
Reads the current mirror partition states from the internal __mirror_state topic on the destination cluster.
...
LastMirroredOffset
Allows destination cluster partition leaders in PREPARING state to query the LMO from the source cluster. If the source cluster has no record of this offset in its internal topic, it returns 0, meaning the log must be truncated to the beginning and mirroring starts from scratch. This is particularly important during failback. The last mirrored offset identifies where mirrored data ends and un-mirrored data begins. Records beyond this offset must be truncated before mirroring new data from the new source cluster; otherwise, the two clusters would contain inconsistent data.
ListMirroredOffsetsRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "ReadMirrorStatesRequestLastMirroredOffsetsRequest",
// 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." }
]}
]}
]
} |
...
LastMirroredOffsetsResponse
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "ReadMirrorStatesResponseLastMirroredOffsetsResponse",
// 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": "TopicsErrorMessage", "type": "[]TopicResultstring", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The readtop-level resultserror formessage, the topics.", "fields": [
or null if there was no error." },
{ "name": "NameTopics", "type": "string[]TopicResult", "versions": "0", "entityType": "topicName",
"about": "The results for topicthe nametopics." },, "fields": [
{ "name": "PartitionsName", "type": "[]PartitionResultstring", "versions": "0", "entityType": "topicName",
"about": "The results for the partitionstopic name.", "fields": [ },
{ "name": "PartitionIndexPartitions", "type": "int32[]PartitionResult", "versions": "0",
"about": "The partition indexresults for the partitions." },, "fields": [
{ "name": "LastMirroredOffsetPartitionIndex", "type": "int64int32", "versions": "0",
"about": "The lastpartition mirrored offsetindex." },
{ "name": "StateLastMirroredOffset", "type": "int8int64", "versions": "0+",
"about": "The mirrorlast partitionmirrored stateoffset." },
{ "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.
...
ReadMirrorStatesRequest
| 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 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." }
]}
]}
]
} |
ReadMirrorStatesResponse
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "ReadMirrorStatesResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "LastMirroredOffsetThrottleTimeMs", "type": "int64int32", "versions": "0",
"about": "The last mirrored offset+",
"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": "StateErrorCode", "type": "int8int16", "versions": "0+",
"about": "The mirrorerror partition state." }
]}
]}code, or 0 if there was no error." },
{ "name": "RemovedTopicsErrorMessage", "type": "[]string", "versions": "0+", "aboutnullableVersions": "The topic names to be removed." }
]
} |
WriteMirrorStatesResponse
| Code Block |
|---|
{
"apiKey": TBD0+", "default": "null",
"type": "response",
"nameabout": "WriteMirrorStatesResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
The top-level error message, or null if there was no error." },
{ "name": "ThrottleTimeMsTopics", "type": "int32[]TopicResult", "versions": "0+",
"about": "The durationread inresults milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
the topics.", "fields": [
{ "name": "ErrorCodeName", "type": "int16string", "versions": "0+", "entityType": "topicName",
"about": "The error code, or 0 if there was no errortopic name." },
{ "name": "TopicsPartitions", "type": "[]TopicResultPartitionResult", "versions": "0",
"about": "The write results for the topicspartitions.", "fields": [
{ "name": "NamePartitionIndex", "type": "stringint32", "versions": "0",
"entityType": "topicName",
"about": "The topicpartition nameindex." },
{ "name": "PartitionsLastMirroredOffset", "type": "[]PartitionResultint64", "versions": "0",
"about": "The resultslast formirrored the partitionsoffset.", "fields": [ },
{ "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." }
]}
]}
]
} |
PauseMirrorTopics
WriteMirrorStates
Persists mirror partition state transitions to the internal __mirror_state topic Pauses data replication and metadata sync for the specified mirror topics, keeping them read-only on the destination cluster.
...
WriteMirrorStatesRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "PauseMirrorTopicsRequestWriteMirrorStatesRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "TopicsMirrorName", "type": "[]TopicDatastring", "versions": "0+", "aboutentityType": "The data for the topics.mirrorName",
"fieldsabout": [
"The mirror name." },
{ "name": "TopicIdTopics", "type": "uuid[]TopicData", "versions": "0+",
"about": "The uniquedata for topicthe IDtopics."},
"fields": [
{ "name": "TopicNameName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The topic name." }
]}
]
} |
PauseMirrorTopicsResponse
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"nameentityType": "PauseMirrorTopicsResponsetopicName",
// Version 0 is the initial version.
"validVersionsabout": "0",
The topic name."flexibleVersions": "0+" },
"fields": [
{ "name": "ThrottleTimeMsPartitions", "type": "int32[]PartitionData", "versions": "0+",
"about": "The duration in millisecondsdata for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
partitions.", "fields": [
{ "name": "ErrorCodePartitionIndex", "type": "int16int32", "versions": "0+",
"about": "The error code, or 0 if there was no errorpartition index." },
{ "name": "TopicsLastMirroredOffset", "type": "[]TopicResultint64", "versions": "0",
"about": "The resultslast formirrored the topicsoffset." },
"fields": [
{ "name": "NameState", "type": "stringint8", "versions": "0+",
"entityType": "topicName",
"about": "The mirror topicpartition namestate." },
]}
]},
{ "name": "ErrorCodeRemovedTopics", "type": "int16[]string", "versions": "0+",
"about": "The errortopic code,names orto 0 if there was no errorbe removed." }
]}
]
} |
ResumeMirrorTopics
Resumes data replication and metadata sync for previously paused mirror topics from where they left off.
ResumeMirrorTopicsRequest
WriteMirrorStatesResponse
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],response",
"name": "ResumeMirrorTopicsRequestWriteMirrorStatesResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
"fields": [
{ "name": "TopicIdThrottleTimeMs", "type": "uuidint32", "versions": "0+",
"about": "The unique topic ID."},
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": "TopicNameErrorCode", "type": "stringint16", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The topic name." }
]}
]
} |
ResumeMirrorTopicsResponse
| Code Block |
|---|
{
"apiKey": 103,
error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "responsestring",
"nameversions": "ResumeMirrorTopicsResponse0+",
// Version 0 is the initial version.
"validVersions"nullableVersions": "0+", "default": "0null",
"flexibleVersionsabout": "0+",
"fields": [
The top-level error message, or null if there was no error." },
{ "name": "ThrottleTimeMsTopics", "type": "int32[]TopicResult", "versions": "0+",
"about": "The durationwrite inresults milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
the topics.", "fields": [
{ "name": "ErrorCodeName", "type": "int16string", "versions": "0+", "entityType": "topicName",
"about": "The error code, or 0 if there was no errortopic name." },
{ "name": "TopicsPartitions", "type": "[]TopicResultPartitionResult", "versions": "0",
"about": "The results for the topicspartitions.", "fields": [
{ "name": "NamePartitionIndex", "type": "stringint32", "versions": "0",
"entityType": "topicName",
"about": "The topicpartition nameindex." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
FindCoordinatorRequest
...