Versions Compared

Old Version 210

changes.mady.by.user Federico Valeri

Saved on

compared with

New Version 211

changes.mady.by.user Federico Valeri

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

  1. Connection Management: The manager maintains a connection pool with one blocking sender per source cluster. These connections are created lazily when the first topic for a mirror is added. Each sender uses the security credentials and network settings from the mirror configuration, allowing different mirrors to use different authentication mechanisms.
  2. Topic Metadata Synchronization: Every refresh cycle, the manager fetches topic metadata from source clusters using standard MetadataRequest calls. For each topic in the mirror configuration:
    1. Topic Creation: If a topic exists in the source but not the destination, the manager sends a CreateTopics request to the controller with identical partition count and configurations.
    2. Partition Expansion: If the source topic has more partitions than the destination, the manager sends a CreatePartitions request to scale up the destination topic to match.
    3. Configuration Sync: Topic configurations are compared between source and destination. Any differences trigger an IncrementalAlterConfigs request to align destination configs with the source.
    4. Topic Auto-Discovery: Periodically discovers new topics on the source cluster that match mirror.topics.include and do not match mirror.topics.exclude, sending a StartMirrorTopicsRequest to the controller for atomic topic creation. Exclude patterns are also enforced on already-mirroring topics.
    5. Topic Deletion: When a topic is deleted on the source cluster, the mirror partitions on the destination cluster moves to STOPPED state. This prevents accidental deletions to affect the destination cluster. In case it was intentional, the operator would need to manually remove the topic from the mirror.
  3. Consumer Group Offset Synchronization: The manager synchronizes classic and share consumer group offsets to enable seamless failover (no offset translation):
    1. Consumer groups: Lists all consumer groups using ListGroups request.
    2. Fetches committed offsets for each group using OffsetFetch request or DescribeShareGroupOffsets request.
    3. on the source cluster, then fetches their committed offsets. For each group, offsets are filtered to include only partitions belonging to actively mirrored topics. Groups that are active on the destination cluster (i.e., in any state other than EMPTY or DEAD) are skipped to avoid regressing offsets after failover. The filtered offsets are committed to the destination.
    4. Share groups: Lists share groups on the source cluster, then fetches their Share‐Partition Start Offsets (SPSO). Offsets are filtered and active‐group checks are applied in the same way as for consumer groups. The filtered offsets are committed to the destinationCommits those offsets to the destination cluster's group coordinator using the internal OffsetCommit or AlterShareGroupOffsets request.
  4. ACL Synchronization: Access control lists are mirrored from source to destination to maintain consistent security policies:
    1. Fetches all ACLs from the source using DescribeAcls request.
    2. Compares with the destination cluster's current ACLs from the metadata image.
    3. Creates missing ACLs using CreateAcls request.
    4. Deletes ACLs that exist in destination but not in source using DeleteAcls request.

...

Metadata synchronization operates at the mirror level rather than the partition level, so it uses a separate coordinator assignment based on the mirror name alone. Only the broker assigned as the metadata coordinator for a given mirror performs synchronization, and it applies changes only to the mirror partitions it manages. This avoids both redundant synchronization across brokers and unnecessary updates to partitions managed by other coordinators.

Each mirror can define its own filtering rules independently, loaded from the manager at each refresh cycle:

  • The mirror.topic.properties.exclude config controls which topic configuration properties are excluded from synchronization using regex patterns.
    • 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 (default)
    • .*throttled.* (exclude all throttle-related properties)
    • min.insync.replicas,unclean.leader.election.enable (exclude only these two properties)
  • The mirror.groups.include config controls which consumer group offsets are synced using regex patterns.
    • .* (all groups by default)
    • app-.* (only sync groups starting with app-)
    • app-.*,service-.* (sync groups starting with app- or service-)

...

to partitions managed by other coordinators.

MirrorFetcherThread

The MirrorFetcherManager (MFM) extends AbstractFetcherManager to handle fetcher thread lifecycle for mirror partitions. It uses a three-dimensional key (fetcher ID, source broker endpoint, mirror name) to organize threads, ensuring that:

...

Key

Description

Default

bootstrap.servers

A list of host/port pairs to use for establishing the initial connection to 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

mirror.topics.include

A comma-separated list of regex patterns for topic names to include in mirroring. Topics on the source cluster whose names match at least one of the patterns will be automatically discovered and mirrored.

 


mirror.topics.exclude

A comma-separated list of regex patterns for topic names to exclude from mirroring.  Topics matching the exclude pattern are not mirrored even if they match mirror.topics.include. Internal topics are always excluded. Exclude always wins over include. By default internal topics (starting with __ ) are excluded. 

__.*


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.

.*


mirror.groups.exclude

A comma-separated list of regex patterns for consumer group IDs to exclude from offset synchronization.  Groups Groups matching the exclude pattern are not replicated even if they match mirror.groups.include.


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.

Examples:

TOPIC;orders.* (all ACLs for topics matching orders.*)

*;*;*;*;User:alice (all ACLs for principal User:alice)

*;*;*;*;User:app-.* (all ACLs for principals matching User:app-.*)

TOPIC;*;READ;ALLOW (all topic READ/ALLOW ACLs)

GROUP;consumer-.*;READ;ALLOW;User:bob (READ/ALLOW ACLs on groups matching consumer-.* for User:bob)

TOPIC;orders.*,*;*;*;*;User:alice (sync all topic ACLs for orders.* topics and all ACLs for User:alice)

*

security.protocol

Protocol for source cluster communication (PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL).


sasl.*

SASL configuration properties.


ssl.*

SSL configuration properties.


...

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

PreparingPartitionState

MirrorMetadataManager

kafka.server.mirror


Number of partitions in preparing state.

kafka.server.mirror:type=MirrorMetadataManager,name=PreparingPartitionState

...