Versions Compared

Old Version 56

changes.mady.by.user Federico Valeri

Saved on

compared with

New Version 57

changes.mady.by.user Federico Valeri

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.
Comment: Add mirrorName entity type for cluster mirroring RPCs

...

This KIP extends CreateTopic API, but also introduces some new APIs and metadata records.


Introduce a new MIRROR_NAME entity type in the message generator to

  provide schema-level type validation for mirror name fields, consistent                                                                                                                                                                                                                  

  with existing entity types (topicName, groupId, brokerId, etc.).                                                                                                                                                                                                                         

  Annotate all MirrorName fields across the 8 mirroring API schemas with                                                                                                                                                                                                                   

  the new entity type and add corresponding test coverage.

CreateTopic

The CreateTopic API is updated extended to add information required for mirror topic creation.

...

  1. This topic ID is not used by other topics in the current cluster
  2. The replicas for the partition assignment are all active and not in fenced or controlled shutdown. This is to make sure when a topic gets deleted and re-created with the same topic ID, the stale offline log dir won’t be treated as the active log dir after it becomes online (KAFKA-16234).

EntityType

A new entity type is added for  in the message generator to provide schema-level type validation for mirror name fields:

Code Block
public enum EntityType {
    // ... existing types ...
    @JsonProperty("mirrorName")
    MIRROR_NAME(FieldType.StringFieldType.INSTANCE); // New type
}

CreateMirror

The CreateMirror API allows users to create a mirror and supply its configuration. When the broker receives the request, it validates that the mirror name is not already in use, contains only permitted characters, and does not end with the ".removed" suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.

...

Code Block
{ 
  "apiKey": TBD, 
  "type": "request", 
  "listeners": ["broker", "controller"], 
  "name": "CreateMirrorRequest", 
  "validVersions": "0", 
  "flexibleVersions": "0+", 
  "fields": [ 
    { "name": "MirrorName", "type": "string", "versions": "0+", "nullableVersions": "0+", 
       "entityType": "mirrorName", 
      "about": "The cluster mirror name."}, 
    { "name": "Config", "type": "[]MirrorConfig", "versions": "0+", 
      "about": "The cluster mirror configurations.",  "fields": [ 
      { "name": "Name", "type": "string", "versions": "0+", "mapKey": true, 
        "about": "The configuration key name." }, 
      { "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+", 
        "about": "The value to set for the configuration key."} 
    ]} 
  ] 
} 

CreateMirrorResponse

Code Block
{ 
  "apiKey": TBD, 
  "type": "response", 
  "name": "CreateMirrorResponse", 
  "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+", "ignorable": true, 
      "about": "The error message, or null if there was no error." } 

  ] 

}

...

Code Block
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "AddTopicsToMirrorRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]TopicState", "versions": "0+", "about": "The topic state.",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
        { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, 
          "entityType": "topicName",
           "about": "The topic name." },
        { "name": "MirrorName", "type": "string", "versions": "0+", "nullableVersions": "0+", "entityType": "mirrorName",
                    "about": "The cluster mirror name."}
      ]}
  ]
}

...

Code Block
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "LastMirroredOffsetsRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicState", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." }
      ]}
    ]}
  ]
}

...

Code Block
{
  "apiKey": TBD,
  "type": "response",
  "name": "ListMirrorsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
      "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": "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": "SourceBootstrap", "type": "string", "versions": "0+",
        "about": "The source cluster bootstrap servers." },
      { "name": "TopicCount", "type": "int32", "versions": "0+", "default": "0",
        "about": "The number of topics configured for this mirror. 0 indicates an empty mirror with no topics." }
    ]}
  ]
}

...

Code Block
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "DescribeMirrorsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorNames", "type": "[]string", "versions": "0+", "entityType": "mirrorName",
       
      "about": "The names of the mirrors to describe. Null or empty array means all mirrors." },
    { "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+", "default": "false",
      "about": "Whether to include authorized operations." }
  ]
}

...

Code Block
{
  "apiKey": TBD,
  "type": "response",
  "name": "DescribeMirrorsResponse",
  // 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": "Mirrors", "type": "[]DescribedMirror", "versions": "0+",
      "about": "Each described mirror.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The error code, or 0 if there was no error." },
      { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
                "about": "The cluster mirror name." },
      { "name": "Topics", "type": "[]TopicPartitions", "versions": "0+",
        "about": "Each topic in the mirror.", "fields": [
        { "name": "TopicName", "type": "string", "versions": "0+",
          "about": "The topic name." },
        { "name": "Partitions", "type": "[]PartitionDetail", "versions": "0+",
          "about": "Each partition detail.", "fields": [
          { "name": "PartitionIndex", "type": "int32", "versions": "0+",
            "about": "The partition index." },
          { "name": "SourceOffset", "type": "int64", "versions": "0+",
            "about": "The high watermark offset from the source cluster leader." },
          { "name": "DestinationOffset", "type": "int64", "versions": "0+",
            "about": "The log end offset on the destination cluster." },
          { "name": "Lag", "type": "int64", "versions": "0+",
            "about": "The lag (source offset - destination offset)." },
          { "name": "State", "type": "string", "versions": "0+",
            "about": "The partition state (INITIALIZING, PREPARING, MIRRORING, STOPPING, STOPPED, FAILED)." }
        ]}
      ]},
      { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
        "about": "32-bit bitfield to represent authorized operations for this mirror." }
    ]}
  ]
}

...

Code Block
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "ReadMirrorStatesRequest",
  "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." }
        ]}
      ]}
  ]
}

...

Code Block
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "WriteMirrorStatesRequest",
  "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." },
        { "name": "LastMirroredOffset", "type": "int64", "versions": "0",
          "about": "The last mirrored offset." },
        { "name": "State", "type": "int8", "versions": "0+",
          "about": "The mirror partition state." }
      ]}
    ]},
    { "name": "RemovedTopics", "type": "[]string", "versions": "0+", "about": "The topic names to be removed." }
  ]
}

...