Page History

...

The MirrorMetadataManager (MMM) implements periodic metadata synchronization between source and destination clusters. It maintains persistent network connections to all source clusters. During periodic metadata refresh, the broker validates that the source cluster ID has not changed. If a mismatch is detected, metadata sync for that mirror is halted and an error is logged. This prevents silent data corruption in case of misconfiguration or unintended source cluster replacement.

Responsibilities:

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 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. Lists all consumer groups using ListGroups request.
2. Fetches committed offsets for each group using OffsetFetch request or DescribeShareGroupOffsets request.
3. Commits 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.

...

An operator can grant ClusterMirror:*:CREATE,ALTER,DESCRIBE for full mirror management, or scope it to specific mirrors like ClusterMirror:prod-dr:DESCRIBE for read-only monitoring of a single mirror, without granting any broker-level privileges.

Inter-broker coordinator RPCs (WriteMirrorStates, ReadMirrorStates, LastMirroredOffsets, FindCoordinator with CoordinatorType.MIRROR) require CLUSTER_ACTION on the Cluster resource, as they are only issued by the broker service account. MMM issues

CreatePartitions, OffsetCommit, IncrementalAlterConfigs, CreateAcls, DeleteAcls locally using and DeleteAcls are issued internally by MMM through the inter-broker principalchannel or direct coordinator calls, bypassing normal ACL checks. This is by design but means the mirror feature implicitly holds ALTER on topics, groups, and ACLs within the destination clusterNo explicit ACL grants are needed for these operations.

Idempotent Producer

The idempotent producers rely on producer IDs to detect duplicate writes and ensure idempotent production. To avoid conflicts with the destination cluster's producer ID space, we rewrite source producer IDs to occupy the unused negative space by applying the formula: 

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "ListMirrorsResponse",
  // 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+", "default": "null",
      "about": "The top-level error message, or null 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": "TopicCountSourceBootstrap", "type": "int32string", "versions": "0+", "default": "0",
        "about": "The numbersource ofcluster topics configured for this mirror. 0 indicates an empty mirror with no topics.bootstrap servers." },
      { "name": "ClusterIdSourceClusterId", "type": "string", "versions": "0+", "default": "",
        "about": "The source cluster ID, or empty if not yet resolved." },
      { "name": "BootstrapServerTopicCount", "type": "stringint32", "versions": "0+", "default": "0",
        "about": "The source cluster bootstrap servers number of topics configured for this mirror. 0 indicates an empty mirror with no topics." }
    ]}
  ]
}

DescribeMirrors

...