DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
Consider this source cluster log:
Offset | Type | isTxn | PID | Content |
0 | DATA_RECORD | true | 4001 | key=A, value=1 |
1 | DATA_RECORD | true | 4001 | key=B, value=2 |
2 | DATA_RECORD | true | 4002 | key=X, value=9 |
3 | CONTROL_MARKER | true | 4001 | COMMIT marker for PID 4001 |
4 | CONTROL_MARKER | true | 4002 | ABORT marker for PID 4002 |
5 | DATA_RECORD | false | none | key=Z, value=10 |
If replication reaches offset 4 and the source cluster fails, the destination cluster contains data records for transaction 4002 (offset 2) without the abort marker (offset 4). This creates a hanging transaction that can never be committed or aborted on the destination cluster.
...
Cluster mirrors can be configured using the following properties:
Topic Configuration
Key | Description | Default | Dynamic |
|---|---|---|---|
mirror.name | 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 |
mirror.metadata.refresh.interval.ms | 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.ms | Request timeout for source cluster communication. | 30000 | |
socket.connection.setup.timeout.ms | Socket connection setup timeout. | 10000 | |
reconnect.backoff.ms | Backoff time before reconnection attempts. | 50 | |
send.buffer.bytes | TCP send buffer size. | 131072 | |
receive.buffer.bytes | TCP receive buffer size. | 65536 | |
replica.fetch.backoff.ms | 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). | ||
replica.fetch.wait.max.ms | 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). | ||
replica.socket.timeout.ms | 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. | ||
sasl.kerberos.service.name | 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). | ||
sasl.oauthbearer.scope.claim.name | OAuth scope claim name for token requests. | ||
sasl.oauthbearer.sub.claim.name | 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.
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
...
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. |
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:
...
- Performance Benchmark: Measure replication throughput and latency across WAN.
- Scalability Test: Replicate 1000 topics with 100,000 partitions across clusters.
- Failover Test: Simulate source cluster failure, measure consumer recovery time.
- Long-Running Stability: Run continuous replication for 7 days, verify no memory leaks or performance degradation.
- Security Validation: Test all authentication mechanisms (SASL PLAIN, SCRAM, Kerberos, mTLS) via kafka-mirrors.sh config files.
Rejected Alternatives
Keep
...
Using MirrorMaker 2:
This KIP introduces native cluster mirroring to address the limitations of MirrorMaker 2 described in the motivation section.
Support
...
Unclean Leader Election
As described in the non-goal section, since there's no shared leader epoch between source and detination destination cluster, supporting unclean leader election becomes very tricky.
For example:
source cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
Suppose we mirror everything from the source into destination cluster, including the leader epoch in batches:
target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
This could happen:
- leadership Leadership change in the source cluster, bumping the leader epoch to 2.
- New records appended to the source cluster: offset=2, epoch=2, value=C.
- Before target cluster fetch from the source to identify the leader epoch update, source cluster down, failover happened.
- Topics in destination cluster becomes become writable, new records appended from producer: offset=2, epoch=2, value=DKeep using MirrorMaker 2:
This KIP is to address the drawbacks existing MirrorMaker 2 as described in the motivation section.
Inconsistent result:
source cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
offset 2, epoch: 2, value: C
target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
offset 2, epoch: 2, value: D
The above issue above can be resolved by the LastMirroredOffset API we did propose in this KIP.
The flow will be like this:
- Leadership change in the source cluster, bumping the leader epoch to 2.
- New records appended to the source cluster: offset=2, epoch=2, value=C.
- Before target cluster fetch from the source to identify the update, source cluster down, failover happenedhappens.
- When failover, the destination cluster will store the current last mirrored offset (1 in this case) into internal topic.
- Topics in destination cluster becomes writable, new records appended from producer: offset=2, epoch=2, value=D.
- When the old source cluster wants to reverse mirroring to the new source cluster, it'll firstly ask for the last mirrored offset
,which is 1 in this case. Then, truncate data to offset 1. - Then, start fetch from offset 1.
...
- fetc
As described in the non-goal section, since there's no shared leader epoch between source and detination cluster, supporting unclean leader election becomes very tricky.
For example:
source cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
Suppose we mirror everything from the source into destination cluster, including the leader epoch in batches:
target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
This could happen:- Leadership change in the source cluster, bumping the leader epoch to 2.
- New records appended to source cluster: offset=2, epoch=2, value=C.
- Before target cluster fetch from the source to identify the leader epoch update, source cluster down, failover happened.
- Topics in destination cluster becomes writable, new records appended from producer: offset=2, epoch=2, value=D.
Inconsistent result:
source cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
offset 2, epoch: 2, value: C
target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
offset 2, epoch: 2, value: DThe above issue can be resolved by the LastMirroredOffset API we propose in this KIP.
The flow will be like this:
- Leadership change in the source cluster, bumping the leader epoch to 2.
- New records appended to the source cluster: offset=2, epoch=2, value=C.
- Before target cluster fetch from the source to identify the update, source cluster down, failover happened.
- When failover, the destination cluster will store the current last mirrored offset (1 in this case) into internal topic.
- Topics in destination cluster becomes writable, new records appended from producer: offset=2, epoch=2, value=D.
- When the old source cluster wants to failback to the new source cluster, it'll firstly ask for the last mirrored offset
,which is 1 in this case. Then, truncate data to offset 1. - Then, start fetching from offset 1.
It works well, but when unclean leader election comes into the play, it'll become complicated:
- Unclean leader election happened and leadership change in the source cluster, bumping the leader epoch to 2.
- New leader has empty log in disk.
- New records appended to the source cluster: offset=0, epoch=2, value=C
. - Before target cluster fetch from the source to identify the update, source cluster down, failover happened.
- When failover, the destination cluster will store the current last mirrored offset (1 in this case) into internal topic.
- Topics in destination cluster becomes writable, new records appended from producer: offset=2, epoch=2, value=D
. - When the old source cluster wants to failback to the new source cluster, it'll firstly ask for the last mirrored offset
,which is 1 in this case. Then, truncate data to offset 1.
After this truncation, the data diverge still exist:
source cluster
leader for foo-0 contains this data:
offset 0, epoch: 2, value: C
target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
offset 2, epoch: 2, value: DIn summary, because mirroring relies on asynchronous fetch and metadata requests, the destination cluster can always miss leadership changes on the source cluster. If a failover occurs while the destination cluster has missed such updates, it cannot reconcile its state with the source cluster. This leads to inconsistent data once the old source cluster begins reverse mirroring from the old destination (now the new source). Resolving this requires a shared leader epoch mechanism, which is out of scope for this KIP.
h from offset 1.
It works well, but when unclean leader election comes into the play, it'll become complicated:
- Unclean leader election happened and leadership change in the source cluster, bumping the leader epoch to 2.
- The new leader has an empty log in the disk.
- New records appended to source cluster: offset=0, epoch=2, value=C
. - Unclean leader election happened and leadership change in the source cluster, bumping the leader epoch to 2.
- New leader has empty log in disk.
- New records appended to source cluster: offset=0, epoch=2, value=C
. - Before target cluster fetch from the source to identify the update, source cluster down, failover happened.
- When failover, the destination cluster will store the current last mirrored offset (1 in this case) into internal topic.
- Topics in destination cluster becomes writable, new records appended from producer: offset=2, epoch=2, value=D
. - When the old source cluster wants to reverse mirroring to the new source cluster, it'll firstly ask for the last mirrored offset
,which is 1 in this case. Then, truncate data to offset 1.
After this truncation, the data diverge still exist:
source cluster
leader for foo-0 contains this data:
offset 0, epoch: 2, value: C
target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
offset 2, epoch: 2, value: D
In summary, no matter we store the source partition leader epoch in the target cluster or not, there will always be a gap in the target cluster given it's using async fetch request/response or metadata request/response to get the metadata update. When the target cluster misses some leadership change update and failover to the target clsuter, there is no way to sync up with the source cluster anymore. Thus, the inconsistent data will happen after the old source cluster starts to reverse mirror from the old target cluster (new source). To fix this issue, a shared leader epoch mechanism is required. But that's out of the scope of this KIP. because mirroring relies on asynchronous fetch and metadata requests, the destination cluster can always miss leadership changes on the source cluster. If a failover occurs while the destination cluster has missed such updates, it cannot reconcile its state with the source cluster. This leads to inconsistent data once the old source cluster begins reverse mirroring from the old destination (now the new source). Resolving this requires a shared leader epoch mechanism, which is out of scope for this KIP.