DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
Describe configured mirrors to check their lag compared to their source topics:
| 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 |
Remove a specific topic or set of topics from a mirror (stop mirroring / failover):
| Code Block |
|---|
$ bin/kafka-mirror.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 (stop mirroring / promotion):
...
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. |
Admin Client
New methods are added to the Admin interface for programmatic cluster mirror management, along with their supporting classes:
| Code Block |
|---|
CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options); |
...
AddTopicsToMirrorResult addTopicsToMirror(Map<String, String> topicToMirrorName, AddTopicsToMirrorOptions options); |
...
RemoveTopicsFromMirrorResult removeTopicsFromMirror(String mirrorName, Set<String> topics, RemoveTopicsFromMirrorOptions options); |
...
ListMirrorsResult listMirrors(ListMirrorsOptions options); |
...
DescribeMirrorsResult describeMirrors(Collection<String> mirrorNames, DescribeMirrorsOptions options); |
Protocol Changes
This KIP extends CreateTopic API, but also introduces some new APIs and metadata records.
CreateTopic
The CreateTopic API is updated to add information required for mirror topic creation.
{ "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
{
"apiKey": 94,
"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."}
]}
]
}
{
"apiKey": 94,
"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
{
"apiKey":95,
"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."}
]}
]
}
{
"apiKey":95,
"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
{
"apiKey":96,
"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." }
]}
]
}
{
"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
{
"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." }
]}
]}
]
}
{
"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
{
"apiKey": 98,
"type": "request",
"listeners": ["broker"],
"name": "ListMirrorsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": []
}
{
"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
{
"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." }
]
}
{
"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
{
"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." }
]
}
{
"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
{
"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." }
]
}
{
"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:
public enum CoordinatorType {
GROUP((byte) 0),
TRANSACTION((byte) 1),
MIRROR((byte) 2); // New type
}
Mirror Metadata Records
LastMirroredOffsets
LastMirroredOffsets record tracks the latest successfully mirrored offset for each partition.
{
"apiKey": 1,
"type": "coordinator-key",
"name": "LastMirroredOffsetsKey",
"validVersions": "0",
"flexibleVersions": "none",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0",
"about": "The cluster mirror name."}
]
}
{
"apiKey": 1,
"type": "coordinator-value",
"name": "LastMirroredOffsetsValue",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Topics", "type": "[]Topic", "versions": "0+",
"about": "The mirror topics for which we want to store the last mirrored offsets.", "fields": [
{ "name": "Name", "type": "string", "versions": "0",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]Partition", "versions": "0+",
"about": "Each partition to record the last mirrored offsets.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "LastMirroredOffset", "type": "int64", "versions": "0+",
"about": "The last mirrored offset for this partition." }
]}
]}
]
}
MirrorPartitionState
MirrorPartitionState record represents the lifecycle states of a mirrored partition.
{
"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." }
]
}
Configuration
A new configuration resource type is added for cluster mirrors, which is stored in the cluster metadata internal log:
public enum Type {
// ... existing types ...
MIRROR((byte) 64, "mirror"); // New type
}
Cluster mirrors can be configured using the following properties:
Topic Configuration
Key | Description | Default | Dynamic |
Identifies the mirror that manages this topic. Topics with this configuration set are read-only and can only be modified through mirror management APIs. | “” | yes | |
mirror.replication.throttled.replicas | A list of replicas for which log replication should be throttled on the mirror follower node. The list should describe a set of replicas in the form [PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... or alternatively the wildcard '*' can be used to throttle all replicas for this topic." | MAX_LONG | yes |
Broker Configuration
Key | Description | Default | Dynamic |
mirror.topic.num.partitions | Number of partitions for __mirror_state internal topic. | 50 | no |
mirror.topic.replication.factor | Replication factor for __mirror_state internal topic. | 3 | no |
mirror.num.replica.fetchers | Number of fetcher threads per mirrored source broker, | 1 | yes |
The interval in milliseconds at which the coordinator refreshes metadata from source clusters. This controls how frequently the coordinator polls source clusters to detect new topics and metadata changes. | 30000 | yes | |
Request timeout for source cluster communication. | 30000 | ||
Socket connection setup timeout. | 10000 | ||
Backoff time before reconnection attempts. | 50 | ||
send.buffer.bytes | TCP send buffer size. | 131072 | |
receive.buffer.bytes | TCP receive buffer size. | 65536 | |
Time to wait before retrying fetch requests after failures (e.g., source leader change). | |||
replica.fetch.max.bytes | Maximum bytes to fetch per partition in a single request to the source cluster. | ||
replica.fetch.min.bytes | Minimum bytes that must be available before the source cluster responds to fetch requests (helps reduce cross-datacenter request frequency for low-throughput topics). | ||
replica.fetch.response.max.bytes | Maximum total bytes across all partitions in a single fetch response from source cluster (important for WAN bandwidth management in cluster mirroring). | ||
Maximum time the source cluster will wait to accumulate replica.fetch.min.bytes before responding (balances latency vs. efficiency for cross-cluster replication). | |||
replica.socket.receive.buffer.bytes | TCP receive buffer size for connections to source cluster brokers (larger values can improve throughput over high-latency WAN links). | ||
Socket timeout for read operations from source cluster (should account for cross-datacenter network latency). | |||
mirror.replication.throttled.rate | A long representing the upper bound (bytes/sec) on replication traffic for mirrored follower node enumerated in the property “mirror.replication.throttled.replicas” (for each topic). This property can be only set dynamically. It is suggested that the limit be kept above 1MB/s for accurate behaviour. | yes |
Mirror Configuration
Key | Description | Default | Dynamic |
bootstrap.servers | List of host/port pairs of the source cluster. | ||
mirror.topic.properties.exclude | A comma-separated list of topic config property names to exclude from synchronization. Properties in this list will not be replicated from the source cluster. The mirror.name property is always excluded regardless of this setting. | follower.replication.throttled.replicas, leader.replication.throttled.replicas,message.timestamp.difference.max.ms,log.message.timestamp.before.max.ms,log.message.timestamp.after.max.ms,message.timestamp.type,unclean.leader.election.enable,min.insync.replicas,mirror.name | yes |
mirror.groups.include | A comma-separated list of regex patterns for consumer group IDs to include in offset synchronization. Only consumer groups whose IDs match at least one of the patterns will have their offsets replicated from the source cluster. | .* | yes |
mirror.acl.include | A comma-separated list of ACL include rules. Each rule uses semicolon-separated fields: resourceType;resourceName;operation;permissionType;principal. Use '*' as wildcard for any field. The resourceName field supports regex patterns. Trailing wildcard fields can be omitted. See AclRule javadoc for examples. | * | yes |
security.protocol | Protocol for source cluster communication (PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL). | ||
sasl.mechanism | SASL mechanism (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI, OAUTHBEARER). | ||
sasl.jaas.config | JAAS login context parameters for authentication. | ||
sasl.client.callback.handler.class | Fully qualified name of SASL client callback handler class. | ||
sasl.login.callback.handler.class | Fully qualified name of SASL login callback handler class. | ||
sasl.login.class | Fully qualified name of class implementing Login interface. | ||
Kerberos principal name for source cluster (when using GSSAPI). | |||
sasl.kerberos.ticket.renew.jitter | Percentage of random jitter added to Kerberos ticket renewal time. | ||
sasl.kerberos.ticket.renew.window.factor | Login thread sleep time until renewal as percentage of ticket lifetime. | ||
sasl.kerberos.min.time.before.relogin | Minimum time before attempting Kerberos credential renewal. | ||
sasl.login.refresh.window.factor | Login refresh thread sleep factor relative to credential lifetime. | ||
sasl.login.refresh.window.jitter | Maximum random jitter relative to credential refresh time. | ||
sasl.login.refresh.min.period.seconds | Minimum time between credential refreshes. | ||
sasl.login.refresh.buffer.seconds | Buffer time before credential expiration to maintain. | ||
sasl.oauthbearer.token.endpoint.url | OAuth token endpoint URL (when using OAUTHBEARER). | ||
OAuth scope claim name for token requests. | |||
OAuth subject claim name for principal identification. | |||
ssl.protocol | SSL protocol version (TLSv1.2, TLSv1.3). | ||
ssl.provider | Name of security provider for SSL connections. | ||
ssl.cipher.suites | List of enabled SSL cipher suites. | ||
ssl.enabled.protocols | List of enabled SSL/TLS protocol versions. | ||
ssl.keystore.type | Keystore file format (JKS, PKCS12, PEM). | ||
ssl.keystore.location | Path to keystore file containing client certificate and private key. | ||
ssl.keystore.password | Password for the keystore file. | ||
ssl.keystore.key | Private key in PEM format (alternative to keystore file). | ||
ssl.keystore.certificate.chain | Certificate chain in PEM format (alternative to keystore file). | ||
ssl.key.password | Password for the private key in the keystore. | ||
ssl.truststore.type | Truststore file format (JKS, PKCS12, PEM). | ||
ssl.truststore.location | Path to truststore file for verifying source cluster broker certificates. | ||
ssl.truststore.password | Path to truststore file for verifying source cluster broker certificates. | ||
ssl.truststore.certificates | Trusted certificates in PEM format (alternative to truststore file). | ||
ssl.keymanager.algorithm | Algorithm used by KeyManager factory (default: SunX509). | ||
ssl.trustmanager.algorithm | Algorithm used by TrustManager factory (default: PKIX). | ||
ssl.endpoint.identification.algorithm | Endpoint identification algorithm for hostname verification (https or empty to disable). | ||
ssl.secure.random.implementation | SecureRandom PRNG implementation for SSL cryptography. | ||
ssl.engine.factory.class | Fully qualified name of class implementing SslEngineFactory for custom SSL engine creation. |
Metrics
A core set of metrics will be provided with the initial implementation.
Metric Name | Type | Group | Tags | Description | JMX Bean |
MaxLag | MirrorFetcherManager | kafka.server.mirror | clientId=MirrorReplica | Max lag in messages between destination leader and source leader replicas. | kafka.server.mirror:type=MirrorFetcherManager,name=MaxLag,clientId=MirrorReplica |
MinFetchRate | MirrorFetcherManager | kafka.server.mirror | clientId=MirrorReplica | The min fetch rate between destination leader and source leader replicas. | kafka.server.mirror:type=MirrorFetcherManager,name=MirrorReplica |
ConsumerLag | FetcherLagMetrics | kafka.server | clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},topic=([-.\w]+),partition=([0-9]+) | Lag in messages per remote leader replica. | kafka.serverr:type=FetcherLagMetrics,name=ConsumerLag,clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},topic=([-.\w]+),partition=([0-9]+) |
DeadThreadCount | MirrorFetcherManager | kafka.server.mirror | clientId=MirrorReplica | Number of dead mirror fetcher threads. | kafka.server,mirror:type=MirrorFetcherManager,name=DeadThreadCount,clientId=MirrorReplica |
FailedPartitionsCount | MirrorFetcherManager | kafka.server.mirror | clientId=MirrorReplica | Total count for failed partitions for any reason like auth, authorization, failed network with source. | kafka.serve.mirrorr:type=MirrorFetcherManager,name=FailedPartitionsCount,clientId=MirrorReplica |
BytesPerSec | FetcherStats | kafka.server | clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},brokerHost={host},brokerPort={port} | Extend kafka.server.FetcherStats to report mirror fetcher threads. | kafka.server:type=FetcherStats,name=BytesPerSec,clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},brokerHost={host},brokerPort={port},mirror-name={mirrorName} |
RequestsPerSec | FetcherStats | kafka.server | MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},brokerHost={host},brokerPort={port} | Extend kafka.server.FetcherStats to report mirror fetcher threads. | kafka.server:type=FetcherStats,name=RequestsPerSec,cclientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName}, brokerHost={host},brokerPort={port},mirror-name={mirrorName} |
[LocalTimeMs,MessageConversionsTimeMs, RemoteTimeMs,RequestBytes, RequestQueueTimeMs,ResponseQueueTimeMs, ResponseSendTimeMs,TemporaryMemoryBytes, TotalTimeMs] | RequestMetrics | kafka.network | request=[mirror_requests] | Extend kafka.network:type=RequestMetrics to list cluster mirror requests. | kafka.network:type=RequestMetrics,name=*, request=* |
ErrorsPerSec | RequestMetrics | kafka.network | request=[mirror_requests],error=* | Extend kafka.network:type=RequestMetrics to list cluster mirror requests. | kafka.network:type=RequestMetrics,name=ErrorsPerSec, request=*, error=* |
RequestsPerSec | RequestMetrics | kafka.network | request=[mirror_requests],version=* | Extend kafka.network:type=RequestMetrics to list cluster mirror requests. | kafka.network:type=RequestMetrics,name=RequestsPerSec, request=*, version=* |
connection-close-rate, connection-close-total, connection-count, connection- creation-rate, connection- creation-total, failed-authentication-rate, failed-authentication-total, failed- reauthentication-rate, failed- reauthentication-total, incoming-byte-rate, incoming-byte-total, network-io-rate, network-io-total, outgoing- byte-rate, outgoing-byte-total, reauthentication-latency-avg, reauthentication-latency-max, request-rate, request-size-avg, request-size-max, request-total, response-rate, response-total, select-rate, select-total, successful-authentication-no- reauth-total, successful- authentication-rate, successful- authentication-total, successful-reauthentication- rate, successful- reauthentication-total | mirror-broker-{DestinationBroker.id}-fetcher-{fetcherId}-mirror-{mirrorName}-metrics | kafka.server | broker-id={sourceBroker.id},fetcher-id={fetcherId} | Fetcher requests in the cluster mirror metrics. | kafka.server:type=mirror-broker-{sourceBroker.id}-fetcher-{fetcherId}-mirror-{mirrorName}-metrics,broker-id={sourceBroker.id},fetcher-id={fetcherId} |
MetadataRefreshError | MirrorMetadataManager | kafka.server.mirror | Number of topic metadata refresh sync errors. | kafka.server.mirror:type=MirrorMetadataManager,name=aclSyncError | |
TopicConfigMetadataSyncError | MirrorMetadataManager | kafka.server.mirror | Number of topic configuration sync errors. | ||
ConsumerGroupOffsetSyncError | MirrorMetadataManager | kafka.server.mirror | Number of CGs sync errors. | ||
AclSyncError | MirrorMetadataManager | kafka.server.mirror | Number of ACLs sync errors. | kafka.server.mirror:type=MirrorMetadataManager,name=aclSyncError | |
byte-rate | MirrorReplication | kafka.server | Bandwidth quota metrics. Indicates the throttled data mirror replication rate of the broker in bytes/sec. | kafka.server:type=MirrorReplication | |
FailedPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in failed state. | kafka.server.mirror:type=MirrorMetadataManager,name=FailedPartitionState | |
StoppedPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in a stopped state. | kafka.server.mirror:type=MirrorMetadataManager,name=StoppedPartitionState | |
StoppingPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in stopping state. | kafka.server.mirror:type=MirrorMetadataManager,name=StoppingPartitionState | |
MirroringPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in mirroring state. | kafka.server.mirror:type=MirrorMetadataManager,name=MirroringPartitionState | |
PreparingPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in preparing state. | kafka.server.mirror:type=MirrorMetadataManager,name=PreparingPartitionState |
Compatibility, Deprecation, and Migration Plan
...