DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
| Code Block |
|---|
public enum EntityType {
// ... existing types ...
@JsonProperty("mirrorName")
MIRROR_NAME(FieldType.StringFieldType.INSTANCE); // New type
} |
CreateMirror
The CreateMirror API allows Allows users to create a mirror and supply its configuration. When the broker receives the request, it validates that the mirror name is not already in use, contains only permitted characters, and does not end with the ".removed" suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.
...
| 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+", "ignorable": true,
"about": "The error message, or null if there was no error." }
]
} |
AddTopicsToMirror
The AddTopicsToMirror API adds 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.
...
| 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": "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." }
]}
]
} |
RemoveTopicsFromMirror
The RemoveTopicsFromMirror API allows 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.
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "RemoveTopicsFromMirrorResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "Topics", "type": "[]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
The LastMirroredOffset API allows 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.
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "LastMirroredOffsetsResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "Topics", "type": "[]TopicResult", "versions": "0",
"about": "The results for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
"about": "The results for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "LastMirroredOffset", "type": "int64", "versions": "0",
"about": "The last mirrored offset." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
ListMirrors
The ListMirrors API returns 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+", "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+", "entityType": "mirrorName",
"about": "The cluster 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
The DescribeMirrors API is to retrieve the information about the mirror names, including the partition state, lag, source offset and destination offset Returns the current mirroring status, state, and configuration for the specified mirror topics on the destination cluster.
DescribeMirrorsRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker"],
"name": "DescribeMirrorsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorNames", "type": "[]string", "versions": "0+", "entityType": "mirrorName",
"about": "The names of the mirrors to describe. Null or empty array means all mirrors." },
{ "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+", "default": "false",
"about": "Whether to include authorized operations." }
]
} |
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "DescribeMirrorsResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "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+",
"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
The ReadMirrorStates RPC reads mirror Reads the current mirror partition states from the coordinator broker when it resides on a different node than the requesting brokerinternal __mirror_state topic on the destination cluster.
ReadMirrorStatesRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "ReadMirrorStatesRequest",
// 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." }
]}
]}
]
} |
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "ReadMirrorStatesResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "Topics", "type": "[]TopicResult", "versions": "0",
"about": "The read results for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
"about": "The results for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "LastMirroredOffset", "type": "int64", "versions": "0",
"about": "The last mirrored 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
The WriteMirrorStates RPC writes Persists mirror partition state updates transitions to the coordinator broker when it resides on a different node than the requesting brokerinternal __mirror_state topic on the destination cluster.
WriteMirrorStatesRequest
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "WriteMirrorStatesRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The mirror name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0",
"about": "The data for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData", "versions": "0",
"about": "The data for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "LastMirroredOffset", "type": "int64", "versions": "0",
"about": "The last mirrored 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." }
]
} |
...
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "WriteMirrorStatesResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "Topics", "type": "[]TopicResult", "versions": "0",
"about": "The write results for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
"about": "The results for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
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": "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." }
]}
]
} |
PauseMirrorTopicsResponse
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "PauseMirrorTopicsResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "Topics", "type": "[]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." }
]}
]
} |
ResumeMirrorStates
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": "ResumeMirrorTopicsRequest",
// 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": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
{ "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The topic name." }
]}
]
} |
ResumeMirrorTopicsResponse
| Code Block |
|---|
{
"apiKey": 103,
"type": "response",
"name": "ResumeMirrorTopicsResponse",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "Topics", "type": "[]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." }
]}
]
} |
FindCoordinatorRequest
...