DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
RPC | Component | ACL Operation | ACL Resource | Purpose |
| Fetch | MFT | Read | Topic | Data replication |
| Metadata | MMM | Describe | Topic | Topic discovery and leader tracking |
| DescribeConfigs | MMM | DescribeConfigs | Topic | Topic configuration sync |
| ListGroups | MMM | Describe | Group | Consumer group offset sync |
| OffsetFetch | MMM | Describe | Group | Consumer group offset sync |
| DescribeAcls | MMM | Describe | Cluster | ACL synchronization |
| DescribeMirrors | MC | Read | Cluster | Log truncation when preparing |
| ApiVersions | MMM | Feature negotiation | ||
| ListOffsets | MFT | Describe | Topic | Offset bounds discovery |
| OffsetsForLeaderEpoch | MFT | Describe | Topic | Leader epoch validation for truncation |
...
Integrating with tiered storage would allow mirroring to handle data that has been offloaded to remote storage (e.g., S3, HDFS), enabling full replication of topics with long retention periods without requiring all data to reside in local broker storage. During the PREPARING LOG_TRUNCATION state, the mirror truncates the local log to the LME. This operation does not support tiered storage on the destination cluster because LME may be moved to remote segments.
When tiered storage is enabled locally for a mirror topic, its partitions transition to FAILED state. This limitation applies only to the destination cluster during PREPARING log truncation phase, and will be removed once tiered storage truncation will be fully supported. A detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP.
...
Before transitioning a mirror partition from PREPARING to LOG_TRUNCATION to MIRRORING, the MirrorCoordinator must ensure that all in-sync replicas in the destination cluster have truncated their logs to the correct offset. If less than min ISR are available, we will skip and retry in the following fetch. This coordination step validates that every ISR member has completed truncation before the partition is allowed to begin actively fetching from the source cluster. Without it, the mirror leader could start appending new data from the source while local followers still hold divergent log segments, causing inconsistencies within the destination cluster. After truncation, reverse mirroring begins normally. Note that the log truncation on reverse mirroring may cause the data loss if there are records that didn't get mirrored to the old destination cluster.
...
- User sends StartMirrorTopicsRequest with mirror name, topics, and optional include/exclude patterns.
- Controller persists include/exclude patterns as ConfigRecord entries on the MIRROR resource in the metadata log.
- For each topic, the controller creates it on the destination if it does not already exist, and sets mirror.name=<mirrorName> on the TOPIC resource config. Both operations are written in a single metadata record batch.
- Brokers receive the metadata update. The MirrorMetadataManager detects the new mirror.name config (without .stopped or .paused suffix) and queries the coordinator for the current partition state.
- Partitions transition from UNKNOWN to PREPARINGLOG_TRUNCATION. During PREPARINGthis state, LME truncation runs and waits for all ISR members (or all replicas if ULE is enabled).
- Partitions transition to MIRRORING. A MirrorFetcherThread is created and begins fetching from the source cluster.
- Partition state is persisted to __mirror_state.
- On subsequent metadata refresh cycles, the MirrorMetadataManager discovers new source topics matching the persisted include/exclude patterns and repeats steps 3-7 for each.
...
Stop mirroring for the specified topics. The broker validates that all target topic partitions are in either PREPARING LOG_TRUNCATION or MIRRORING state. Once validated, the request is forwarded to the controller, which appends the .stopped suffix to the mirror.name topic config to mark the topics as no longer mirrored.
...
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.mirror: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 | clientId=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. | ||
ShareGroupOffsetSyncError | MirrorMetadataManager | kafka.server.mirror | Number of SGs sync errors. | ||
AclSyncError | MirrorMetadataManager | kafka.server.mirror | Number of ACLs sync errors. | kafka.server.mirror:type=MirrorMetadataManager,name=aclSyncError | |
ByteRate | 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 | |
LogTruncationPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in log truncation state. | kafka.server.mirror:type=MirrorMetadataManager,name=LogTruncationPartitionState | |
EpochFencingPartitionStatePreparingPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in preparing epoch fencing state. | kafka.server.mirror:type=MirrorMetadataManager,name=PreparingPartitionStateEpochFencingPartitionState |
Errors
List of protocol-level errors returned by the new RPCs:
...