Page History

...

• State Management: Mirror partition states and control records are stored in __mirror_state internal topic. The coordinator loads these metadata on startup and partition leadership changes.
• Partition Assignment: Mirror partitions are distributed evenly to coordinators across all brokers and allows for horizontal scaling. The number of coordinator partitions is configurable via mirror.topic.num.partitions.
• Leader Election: When a broker becomes the leader for a __mirror_state partition, it loads the mirror metadata for all mirrors assigned to that partition and begins coordinating those mirrors. On resignation, it clears its in-memory state to avoid stale metadata.
• Metadata Refresh Scheduling: The coordinator schedules periodic metadata refresh operations by invoking a metadata manager every 30 seconds by default. This ensures that configuration updates and group offset commits in the source cluster are continuously propagated to the destination cluster. The refresh interval is configurable via mirror.metadata.refresh.interval.ms.
• State Transitions: The coordinator manages asynchronous state transitions for mirror partitions. Each partition is an independent replication unit with its own state. When the coordinator is the leader for a mirror partition, it writes the state updates directly to __mirror_state topic, otherwise it reads and writes the state via new internal RPCs, enabling distributed coordination across the cluster.

Image RemovedImage Added

State descriptions:

• UNKNOWN: The partition has no cached state (broker just became leader, state not loaded yet). Not an explicit API-driven state, just the absence of state.
• PREPARINGLOG_TRUNCATION: The coordinator for this partition detects via onMetadataUpdate that it leads a mirror partition. It fetches last mirror epoch from the source cluster and truncates to align the local log with the source.
• MIRRORING: All ISR set members have completed truncation. A mirror fetcher thread is started to continuously replicate records from the source cluster.
• EPOCH_FENCING: The destination leader epoch needs to be bumped. A BumpLeaderEpochs request is sent to the controller. On success, the partition transitions to MIRRORING.
• PAUSING: Triggered by the pause operation. The system removes fetchers for the affected partitions.
• PAUSED: Fetchers have been removed. The partition stays read-only with no active fetchers and no metadata synchronization. On resume, transitions directly to MIRRORING.
• STOPPING: The mirror fetcher is removed, the leader epoch is bumped, ABORT markers are appended for all ongoing transactions, last mirror epochs are persisted, and a MIRROR_PID_RESET record appended.
• STOPPED: The topic becomes writable on the destination cluster. The mirror fetcher is removed and the read-only flag is cleared.
• FAILED: An error occurred during mirroring or stopping. The operator that wants to restart a failed mirror partition can restart topic mirroring. More sophisticated recovery strategies can be added later with a follow-up KIPcoordinator tries to recover the mirror partition and then fail permanently. The operator can still use the CLI tool or admin API to manually recover.

MirrorMetadataManager

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 synchronization for that mirror is halted and an error is logged. This prevents silent data corruption in case of misconfiguration or unintended source cluster replacement.

...

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "ReadMirrorStatesRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicData", "versions": "0",
      "about": "The data for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionData", "versions": "0",
        "about": "The data for the partitions.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." }
        ]}
      ]}
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "ReadMirrorStatesResponse",
  // 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": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The read results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
        "about": "The results for the partitions.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "LastMirrorEpoch", "type": "int32", "versions": "0", "default": "-1",
          "about": "The last mirrormirrored leader epoch, or -1 if not available." },
            { "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

Internal API that persists mirror partition state transitions to the internal __mirror_state topic on the destination cluster.

{
  "apiKeyname": TBD"PreviousState",
  "type": "requestint8",
  "listenerstaggedVersions": ["broker0+", "controllertag"]: 0,
  "namedefault": "WriteMirrorStatesRequest"16,
  // Version 0 is the initial version.
  "validVersionsabout": "0",
  "flexibleVersions": "0+",
  "fields": [
The mirror partition state before the last transition; UNKNOWN if not recorded." },
        { "name": "MirrorNameRetryAttempt", "type": "int16", "taggedVersions": "0+", "tag": 1, "default": 0,
          "about": "The number of automatic retry attempts while in FAILED state." }
      ]}
    ]}
  ]
}

WriteMirrorStates

Internal API that persists mirror partition state transitions to the internal __mirror_state topic on the destination cluster.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "WriteMirrorStatesRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The mirror name." },
    { "name": "Topics", "type": "[]TopicData", "versions": "0",
      "about": "The data for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionData", "versions": "0",
        "about": "The data for the partitions.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "", "type": "int32", "versions": "0", "default": "-1",
          "about": "The last mirror leader epoch, or -1 if not available." },
        { "name": "State", "type": "int8", "versions": "0+",
          "about": "The mirror partition state." }
      ]}
    ]},
    { "name": "StoppedTopics", "type": "[]string", "versions": "0+", "about": "The topic names to be stopped." }
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "WriteMirrorStatesResponse",
  // 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": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The write results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
        "about": "The results for the partitions.", "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." }
      ]}
    ]}
  ]
}

...

{
  "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." },
    { "name": "PreviousState", "type": "int8",  "versions": "0+", "default": 16,
      "about": "The mirror partition index state before this transition; UNKNOWN if not recorded." },
        { "name": "StateRetryAttempt", "type": "int8int16", "versions": "0+",
       "default": "0",
      "about": "The mirror partition number of automatic retry attempts while in FAILED state." }, 
    ]
}

Type Enumerations

This section lists the new values added to existing Kafka type enumerations to support Cluster Mirroring

...

...