DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
The CreateTopic API is updated to add information required for mirror topic creation.
| Code Block |
|---|
// new added { "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." } |
...
]} |
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).
CreateMirror
CreateMirrorRequest
| Code Block |
|---|
{ |
...
"apiKey": |
...
TBD, "type": "request", |
...
"listeners": ["broker", "controller"], |
...
"name": "CreateMirrorRequest", |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "MirrorName", "type": "string", "versions": "0+", "nullableVersions": "0+", |
...
"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", |
...
"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+", "ignorable": true, |
...
"about": "The error message, or null if there was no error." } |
...
] |
...
} |
AddTopicsToMirror
AddTopicsToMirrorRequest
| Code Block |
|---|
{ |
...
"apiKey": |
...
TBD, |
...
"type": "request", |
...
"listeners": ["broker", "controller"], |
...
"name": "AddTopicsToMirrorRequest", |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "Topics", "type": "[]TopicState", "versions": "0+", "about": "The topic state.", |
...
"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": "MirrorName", "type": "string", "versions": "0+", "nullableVersions": "0+", |
...
"about": "The mirror name."} |
...
]} |
...
] |
...
}
{
...
} |
AddTopicsToMirrorResponse
| Code Block |
|---|
{ "apiKey": |
...
TBD, |
...
"type": "response", |
...
"name": "AddTopicsToMirrorResponse", |
...
"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": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."}, |
...
{ "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+", "ignorable": true, |
...
"about": "The error message, or null if there was no error." } |
...
] |
...
} |
RemoveTopicsFromMirror
{
RemoveTopicsFromMirrorRequest
| Code Block |
|---|
{ "apiKey": |
...
TBD, |
...
"type": "request", |
...
"listeners": ["broker", "controller"], |
...
"name": "RemoveTopicsFromMirrorRequest", |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "MirrorName", "type": "string", "versions": "0+", "ignorable": true, |
...
"about": "The cluster mirror name." }, |
...
{ "name": "Topics", "type": "[]TopicState", "versions": "0+", "about": "The topic state.", |
...
"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." } |
...
]} |
...
] |
...
} |
RemoveTopicsFromMirrorResponse
| Code Block |
|---|
...
{ |
...
"apiKey":96, |
...
"type": "response", |
...
"name": "RemoveTopicsFromMirrorResponse", |
...
"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": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."}, |
...
{ "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+", "ignorable": true, |
...
"about": "The error message, or null if there was no error." } |
...
] |
...
} |
LastMirroredOffset
...
ListMirroredOffsetsRequest
| Code Block |
|---|
{ |
...
"apiKey":97, |
...
"type": "request", |
...
"listeners": ["broker", "controller"], |
...
"name": "LastMirroredOffsetsRequest", |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "MirrorName", "type": "string", "versions": "0+", "about": "The mirror name." }, |
...
{ "name": "Topics", "type": "[]TopicState", "versions": "0", |
...
"about": "The responses per topic.", "fields": [ |
...
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName", |
...
"about": "The topic name." }, |
...
{ "name": "Partitions", "type": "[]PartitionState", "versions": "0", |
...
"about": "The responses per partition.", "fields": [ |
...
{ "name": "PartitionIndex", "type": "int32", "versions": "0", |
...
"about": "The partition index." } |
...
]} |
...
]} |
...
] } |
LastMirroredOffsetsResponse
...
| Code Block |
|---|
{ |
...
"apiKey":97, |
...
"type": "response", |
...
"name": "LastMirroredOffsetsResponse", |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", |
...
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }, |
...
{ "name": "ErrorCode", "type": "int16", "versions": "0+", |
...
"about": "The error code, or 0 if there was no error." }, |
...
{ "name": "Topics", "type": "[]OffsetResponseTopic", "versions": "0", |
...
"about": "The responses per topic.", "fields": [ |
...
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName", |
...
"about": "The topic name." }, |
...
{ "name": "Partitions", "type": "[]OffsetResponsePartition", "versions": "0", |
...
"about": "The responses per partition.", "fields": [ |
...
{ "name": "PartitionIndex", "type": "int32", "versions": "0", |
...
"about": "The partition index." }, |
...
{ "name": "LastMirroredOffset", "type": "int64", "versions": "0", |
...
"about": "The last mirrored record offset." }, |
...
{ "name": "ErrorCode", "type": "int16", "versions": "0", |
...
"about": "The error code, or 0 if there was no error." } |
...
]} |
...
]} |
...
] |
...
} |
ListMirrors
...
ListMirrorsRequest
| Code Block |
|---|
{ "apiKey": 98, |
...
"type": "request", |
...
"listeners": ["broker"], |
...
"name": "ListMirrorsRequest", |
...
// Version 0 is the initial version. |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [] } |
ListMirrorsResponse
...
| Code Block |
|---|
{ |
...
"apiKey": 98, |
...
"type": "response", |
...
"name": "ListMirrorsResponse", |
...
// Version 0 is the initial version. |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true, |
...
"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": "Mirrors", "type": "[]ListedMirror", "versions": "0+", |
...
"about": "Each mirror in the response.", "fields": [ |
...
{ "name": "MirrorName", "type": "string", "versions": "0+", |
...
"about": "The mirror name." }, |
...
{ "name": "SourceBootstrap", "type": "string", "versions": "0+", |
...
"about": "The source cluster bootstrap servers." }, |
...
{ "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." } |
...
]} |
...
] |
...
} |
DescribeMirrors
DescribeMirrorsRequest
| Code Block |
|---|
{ |
...
"apiKey": 99, |
...
"type": "request", |
...
"listeners": ["broker"], |
...
"name": "DescribeMirrorsRequest", |
...
// Version 0 is the initial version. |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "MirrorNames", "type": "[]string", "versions": "0+", |
...
"about": "The names of the mirrors to describe. Null or empty array means all mirrors." }, |
...
{ "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+", "default": "false", |
...
"about": "Whether to include authorized operations." } |
...
] |
...
} |
DescribeMirrorsResponse
| Code Block |
|---|
...
{ |
...
"apiKey": 99, |
...
"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": "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+", |
...
"about": "The 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+", |
...
"about": "The high watermark offset from the source cluster leader." }, |
...
{ "name": "DestinationOffset", "type": "int64", "versions": "0+", |
...
"about": "The log end offset on the destination cluster." }, |
...
{ "name": "Lag", "type": "int64", "versions": "0+", |
...
"about": "The lag (source offset - destination offset)." }, |
...
{ "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
ReadMirrorStatesRequest
| Code Block |
|---|
{ |
...
"apiKey":100, |
...
"type": "request", |
...
"listeners": ["broker", "controller"], |
...
"name": "ReadMirrorStatesRequest", |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "MirrorName", "type": "string", "versions": "0+", "about": "The mirror name." }, |
...
{ "name": "Topics", "type": "[]TopicState", "versions": "0", |
...
"about": "The responses per topic.", "fields": [ |
...
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName", |
...
"about": "The topic name." }, |
...
{ "name": "Partitions", "type": "[]PartitionState", "versions": "0", |
...
"about": "The responses per partition.", "fields": [ |
...
{ "name": "PartitionIndex", "type": "int32", "versions": "0", |
...
"about": "The partition index." } |
...
]} |
...
]}, |
...
{ "name": "NeedPartitionStates", "type": "bool", "versions": "0+", "default": "true", |
...
"about": "Need the partition states or only topics states needed." } |
...
] |
...
} |
ReadMirrorStatesResponse
| Code Block |
|---|
...
{ |
...
"apiKey":100, |
...
"type": "response", |
...
"name": "ReadMirrorStatesResponse", |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", |
...
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }, |
...
{ "name": "ErrorCode", "type": "int16", "versions": "0+", |
...
"about": "The error code, or 0 if there was no error." }, |
...
{ "name": "Topics", "type": "[]TopicState", "versions": "0", |
...
"about": "The responses per topic.", "fields": [ |
...
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName", |
...
"about": "The topic name." }, |
...
{ "name": "Partitions", "type": "[]PartitionState", "versions": "0", |
...
"about": "The responses per partition.", "fields": [ |
...
{ "name": "PartitionIndex", "type": "int32", "versions": "0", |
...
"about": "The partition index." }, |
...
{ "name": "LastMirroredOffset", "type": "int64", "versions": "0", |
...
"about": "The last mirrored record offset." }, |
...
{ "name": "state", "type": "int8", "versions": "0+", |
...
"about": "The mirror partition state." }, |
...
{ "name": "ErrorCode", "type": "int16", "versions": "0", |
...
"about": "The error code, or 0 if there was no error." } |
...
]} |
...
]} |
...
] |
...
} |
WriteMirrorStates
WriteMirrorStatesRequest
| Code Block |
|---|
{ |
...
"apiKey":101, |
...
"type": "request", |
...
"listeners": ["broker", "controller"], |
...
"name": "WriteMirrorStatesRequest", |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "MirrorName", "type": "string", "versions": "0+", "about": "The mirror name." }, |
...
{ "name": "TopicsUpdated", "type": "[]TopicState", "versions": "0", |
...
"about": "The topics to be updated.", "fields": [ |
...
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName", |
...
"about": "The topic name." }, |
...
{ "name": "Partitions", "type": "[]PartitionState", "versions": "0", |
...
"about": "The responses per partition.", "fields": [ |
...
{ "name": "PartitionIndex", "type": "int32", "versions": "0", |
...
"about": "The partition index." }, |
...
{ "name": "LastMirroredOffset", "type": "int64", "versions": "0", |
...
"about": "The last mirrored record offset." }, |
...
{ "name": "state", "type": "int8", "versions": "0+", |
...
"about": "The mirror partition state." } |
...
]} |
...
]}, |
...
{ "name": "RemovedTopics", "type": "[]string", "versions": "0+", "about": "The topic names to be removed." } |
...
] |
...
} |
WriteMirrorStatesResponse
| Code Block |
|---|
...
{ |
...
"apiKey":101, |
...
"type": "response", |
...
"name": "WriteMirrorStatesResponse", |
...
"validVersions": "0", |
...
"flexibleVersions": "0+", |
...
"fields": [ |
...
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", |
...
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }, |
...
{ "name": "ErrorCode", "type": "int16", "versions": "0+", |
...
"about": "The error code, or 0 if there was no error." }, |
...
{ "name": "Topics", "type": "[]TopicState", "versions": "0", |
...
"about": "The responses per topic.", "fields": [ |
...
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName", |
...
"about": "The topic name." }, |
...
{ "name": "Partitions", "type": "[]PartitionState", "versions": "0", |
...
"about": "The responses per partition.", "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." } |
...
]} |
...
]} |
...
] |
...
} |
FindCoordinatorRequest
The FindCoordinatorRequest object is extended to support a new coordinator type:
...
Note that some features require support from the source cluster.
Feature | Source Cluster Requirement | Destination Cluster Requirement | Notes |
Core mirroring and failover | 2.1 | 4.x | Kafka 4 is compatible with old clients versions up to 2.1 included. |
Failback (reverse mirroring) | 4.x | 4.x | Requires last mirrored offset tracking on both sides, otherwise it will fallback and truncate to zero, effectively mirroring from scratch. |
Tiered Storage | 3.0 | 4.y | If the source doesn't support Tiered Storage, mirroring continues but tiered segments won't be synchronized. |
Share Groups | 4.x | 4.y | If the source doesn't support share groups, mirroring continues but share group offsets won't be synchronized. |
Performance
MirrorFetcherThread uses the same fetch protocol optimizations as ReplicaFetcherThread:
...