Versions Compared

Old Version 45

changes.mady.by.user Colin McCabe

Saved on

compared with

New Version 46

changes.mady.by.user Colin McCabe

Saved on

Key

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

...

Configuration NamePossible ValuesDescription
process.roles

null

broker

controller

broker,controller

If this is null (absent) then we are in legacy mode.

Otherwise, we are in KIP-500 mode and this configuration determines what roles this process should play: broker, controller, or both.

controller.listeners

null

a listener nameIf non-null, this must be a comma-separated list of listener names.

A comma-separated list of the names of the listeners used by the KIP-500 controller. This is required if this process is a KIP-500 controller. The legacy controller will not use this configuration

Despite the similar name, note that this is different from the "control plane listener" introduced by KIP-291.  The "control plane listener" is used on brokers, not on controllers.

controller.connect

null

If non-null, this must be a comma-separated list of controller URIs.

A comma-separated list of controller URIs that KIP-500 brokers should connect to on startup.  Required for nodes running

endpoints, in the format:

{controller-id}@{controller-host):{controller-port}

When in KIP-500 mode, each node must have this configuration, in order to find out how to communicate with the controller quorum.

controller.connect.security.idprotocolIf non-null, this must be a security protocol such as PLAINTEXT, SSL, etc.When in KIP-500 mode, this describes the security protocol that the broker should use to connect to the controllers.  This defaults to PLAINTEXT if no value is given.
controller.id

a 32-bit ID

a 32-bit ID

The controller id for this server.  If this is not specified, we will default to the broker.id plus 3000.

registration.heartbeat.interval.ms10002000The length of time between broker heartbeats.
registration.lease.timeout.ms800018000The length of time that a broker lease lasts if no heartbeats are made.

...

Code Block
languagejs
{
  "apiKey": 50,
  "type": "request",
  "name": "BrokerHeartbeatRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "CurrentState", "type": "int8", "versions": "0+",
      "about": "The current state that the broker is in." },
    { "name": "TargetState", "type": "int8", "versions": "0+",
      "about": "The state that the broker wants to reach." },
    { "name": "BrokerId", "type": "int32", "versions": "0+",
      "about": "The broker ID." },
    { "name": "BrokerEpoch", "type": "int64", "versions": "0+", "default": "-1",
      "about": "The broker epoch, or -1 if one has not yet been assigned." },
    { "name": "CurMetadataOffset", "type": "int64", "versions": "0+",
      "about": "The highest metadata offset which the broker has reached." },
    { "name": "Listeners", "type": "[]Listener", "nullableVersions": "0+",
      "about": "The listeners of this broker", "versions": "0+", "fields": [
        { "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
          "about": "The name of the endpoint." },
        { "name": "Host", "type": "string", "versions": "0+",
          "about": "The hostname." },
        { "name": "Port", "type": "int16", "versions": "0+",
          "about": "The port." },
        { "name": "SecurityProtocol", "type": "int16", "versions": "0+",
          "about": "The security protocol." }
      ]
    }
  ]
}

{
  "apiKeyname": 50"Features",
  "type": "response[]Feature",
  "namenullableVersions": "BrokerHeartbeatResponse0+",
   "validVersions   "about": "0The features on this broker",
  "flexibleVersionsversions": "0+",
  "fields": [
        { "name": "ThrottleTimeMsName", "type": "int32string", "versions": "0+",
      "aboutmapKey": "Durationtrue,
 in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
 "about": "The feature name." }
        { "name": "ErrorCodeMinSupportedVersion", "type": "int16", "versions": "0+",
          "about": "The errorminimum code,supported orfeature 0 if there was no error." },
level." },
         { "name": "ActiveControllerIdMaxSupportedVersion", "type": "int32int16", "versions": "0+",
          "about": "The ID of the active controller, or -1 if the controller doesn't know." },
    { maximum supported feature level." }
      ]
    }
  ]
}

{
  "apiKey": 50,
  "type": "response",
  "name": "NextStateBrokerHeartbeatResponse",
  "typevalidVersions": "int80",
  "versionsflexibleVersions": "0+",
  "fields": [
    { "aboutname": "The state to which the broker should transition." },
    { "name": "BrokerEpoch", "ThrottleTimeMs", "type": "int64int32", "versions": "0+",
      "defaultabout": "-1",
      "about": "The broker's assigned epochDuration in milliseconds for which the request was throttled due to a quota violation, or -1zero if none was assigned the request did not violate any quota." },
    { "name": "LeaseDurationMsErrorCode", "type": "int64int16", "versions": "0+",
      "about": "IfThe BrokerEpocherror is code, or 0 if there was no error." },
    { "name": "ActiveControllerId", "type": "int32", "versions": "0+",
      "about": "The ID of the active controller, or -1 if the controller doesn't know." },
    { "name": "NextState", "type": "int8", "versions": "0+",
      "about": "The state to which the broker should transition." },
    { "name": "BrokerEpoch", "type": "int64", "versions": "0+", "default": "-1",
      "about": "The broker's assigned epoch, or -1 if none was assigned." },
    { "name": "LeaseDurationMs", "type": "int64", "versions": "0+",
      "about": "If BrokerEpoch is not -1, the number of milliseconds that we want the lease to last in monotonic nanoseconds." }
  ]
}

enum BrokerState {
    UNKNOWN(0),
    INITIAL(1),
    FENCED(2),
    ACTIVE(3),
    SHUTDOWN(4);
}

...

Code Block
{
  "apiKey": 0,
  "type": "metadata",
  "name": "BrokerRecord",
  "validVersions": "0",
  "fields": [ "fields": [
	{ "name": "Id", "type": "int32", "versions": "0+",
	  "about": "The broker id." },
	{ "name": "IdEpoch", "type": "int32int64", "versions": "0+",
	  "about": "The broker idepoch." },
	{ "name": "EpochEndPoints", "type": "int64[]BrokerEndpoint", "versions": "0+",
	  "about": "The endpoints that can be used to communicate with this broker epoch." },
.", "fields": [
		{ "name": "EndPointsName", "type": "[]BrokerEndpointstring", "versions": "0+", "mapKey": true,
		  "about": "The name of endpointsthe that can be used to communicate with this broker.", "fields": [endpoint." },
		{ "name": "Host", "type": "string", "versions": "0+",
		  "about": "The hostname." },
		{ "name": "NamePort", "type": "stringint16", "versions": "0+", "mapKey": true,
		  "about": "The name of the endpointport." },
		{ "name": "HostSecurityProtocol", "type": "stringint16", "versions": "0+",
		  "about": "The security hostnameprotocol." }
	]},
		{ "name": "PortRack", "type": "int16string", "versions": "0+", "nullableVersions": "0+",
		  "about": "The broker rack." }
  ]
}

TopicRecord

Code Block
{
  "aboutapiKey": "The port." }1,
		{  "nametype": "SecurityProtocolmetadata",
  "typename": "int16TopicRecord",
  "versionsvalidVersions": "0+",
		  "aboutfields": "The security protocol." }
	]},
	 [
        { "name": "RackName", "type": "string", "versions": "0+", "nullableVersions": "0+",
	
          "about": "The brokertopic rackname." },
    ]
}

TopicRecord

Code Block
{
  "apiKey    { "name": 1"TopicId",
  "type": "metadatauuid",
  "nameversions": "TopicRecord0+",
   "validVersions       "about": "0",
  "fields": [The unique ID of this topic." },
        { "name": "NameDeleting", "type": "stringboolean", "versions": "0+",
          "about": "TheTrue if this topic name." },
        { "name": "TopicId", is in the process of being deleted." }
  ]
}

PartitionRecord

Code Block
{
  "apiKey": 2,
  "type": "uuidmetadata",
  "versionsname": "0+PartitionRecord",
          "about"validVersions": "The unique ID of this topic." }0",
  "fields": [
     { "name": "DeletingPartitionId", "type": "booleanint32", "versions": "0+",
   "default": "-1",
       "about": "TrueThe if this topic is in the process of being deletedpartition id." },
    ]
}

PartitionRecord

Code Block
{
  "apiKeyname": 2"TopicId",
  "type": "metadatauuid",
  "nameversions": "PartitionRecord0+",
    "validVersions": "0",
  "fields": [  "about": "The unique ID of this topic." },
    { "name": "PartitionIdReplicas", "type":  "[]int32", "versions":  "0+", "default": "-1",
      "about": "The replicas of this partition id, sorted by preferred order." },
    { "name": "TopicIdIsr", "type":  "uuid[]int32", "versions":  "0+",
      "about": "The uniquein-sync IDreplicas of this topic.partition" },
    { "name": "ReplicasRemovingReplicas", "type":  "[]int32", "versions":  "0+",
      "about": "The replicas ofthat we thisare partition,in sortedthe byprocess preferredof orderremoving." },
    { "name": "IsrAddingReplicas", "type":  "[]int32", "versions":  "0+",
      "about": "The replicas that we are in-sync the replicasprocess of this partitionadding." },
    { "name": "RemovingReplicasLeader", "type":  "[]int32", "versions":  "0+", "default": "-1",
      "about": "The replicaslead thatreplica, weor are-1 inif thethere processis ofno removingleader." },
    { "name": "AddingReplicasLeaderEpoch", "type":  "[]int32", "versions":  "0+", "default": "-1",
      "about": "TheAn replicasepoch that wegets incremented areeach intime thewe processchange ofthe addingleader." },

  ]
}

ConfigRecord

Code Block
{         
 { "nameapiKey": "Leader",3,
  "type": "int32metadata",
  "versionsname": "0+ConfigRecord",
  "defaultvalidVersions": "-10",
      "aboutfields": "The lead replica, or -1 if there is no leader." },[
    { "name": "LeaderEpochResourceType", "type": "int32int8", "versions": "0+", "default": "-1",
      "about": "An epoch that gets incremented each time we change the leader." }
  ]
}

ConfigRecord

Code Block
{         
  "apiKey": 3,
  ": "The type of resource this configuration applies to." },
    { "name": "ResourceName", "type": "metadatastring",
  "nameversions": "ConfigRecord0+",
      "validVersionsabout": "0",
  "fields": [The name of the resource this configuration applies to." },         
    { "name": "ResourceTypeName", "type": "int8string", "versions": "0+",
      "about": "The typename of resource this configuration applies to." }, the configuration key." },                  
    { "name": "ResourceNameValue", "type": "string", "versions": "0+",     
      "about": "The namevalue of the resource this configuration applies to." },configuration." }
  ]           
}

IsrChange

Code Block
{
  "apiKey": 4,
 { "nametype": "Namemetadata",
  "typename": "stringIsrChangeRecord",
  "versionsvalidVersions": "0+",
  "fields": [
    { "aboutname": "PartitionId"The name of the configuration key." },, "type": "int32", "versions": "0+", "default": "-1",
      "about": "The partition          id." },
    { "name": "ValueTopicId", "type": "stringuuid", "versions": "0+",     
      "about": "The unique valueID of thethis configurationtopic." },
  ]           
}

IsrChange

Code Block
{
  "apiKeyname": 4"Isr",
  "type":  "metadata[]int32",
  "nameversions":  "IsrChangeRecord0+",
      "validVersionsabout": "0",
  "fields": [The in-sync replicas of this partition" },
    { "name": "PartitionIdLeader", "type": "int32", "versions": "0+", "default": "-1",
      "about": "The partition id lead replica, or -1 if there is no leader." },
    { "name": "TopicIdLeaderEpoch", "type": "uuidint32", "versions": "0+", "default": "-1",
      "about": "The unique ID of this topicAn epoch that gets incremented each time we change the leader." }
  ]
}

AccessControlRecord

Code Block
{,
  "apiKey": 5,
 { "nametype": "Isrmetadata",
  "typename":  "[]int32AccessControlRecord",
  "versionsvalidVersions":  "0+",
      "aboutfields": "The in-sync replicas of this partition" },[
    { "name": "LeaderResourceType", "type": "int32int8", "versions": "0+", "default": "-1",
      "about": "The lead replica, or -1 if there is no leader.resource type" },
    { "name": "LeaderEpochResourceName", "type": "int32string", "versions": "0+", "defaultnullableVersions": "-10+",
      "about": "AnThe epochresource thatname, getsor incrementednull eachif timethis weis changefor the default leaderresource." },
  ]
}

AccessControlRecord

Code Block
{
   { "apiKeyname": 5"PatternType",
  "type": "metadataint8",
  "nameversions": "AccessControlRecord0+",
      "validVersionsabout": "0",
  "fields": [The pattern type (literal, prefixed, etc.)" },
    { "name": "ResourceTypePrincipal", "type": "int8string", "versions": "0+",
      "about": "The resourceprincipal typename." },
    { "name": "ResourceNameHost", "type": "string", "versions": "0+", "nullableVersions": "0+",
      "about": "The resource name, or null if this is for the default resourcehost." },
    { "name": "PatternTypeOperation", "type": "int8", "versions": "0+",
      "about": "The pattern type (literal, prefixed, etc.)operation type." },
    { "name": "PrincipalPermissionType", "type": "stringint8", "versions": "0+",
      "about": "The principal name permission type (allow, deny)." },
  ]
}

FenceBroker

Code Block
{
 { "nameapiKey": "Host"6,
  "type": "stringmetadata",
  "versionsname": "0+FenceBrokerRecord",
      "aboutvalidVersions": "0"The host." },,
  "fields": [
    { "name": "OperationId", "type": "int8int32", "versions": "0+",
      "about": "The operation type broker ID to fence. It will be removed from all ISRs." },
    { "name": "PermissionTypeEpoch", "type": "int8int64", "versions": "0+",
      "about": "The epoch of permissionthe typebroker (allow, deny)to fence." }
  ]
}

...

DeleteTopic

Code Block
{
  "apiKey": 67,
  "type": "metadata",
  "name": "FenceBrokerRecordDeleteTopicRecord",
  "validVersions": "0",
  "fields": [
    { "name": "Id", "type": "int32uuid", "versions": "0+",
      "about": "The brokertopic ID to fencedelete. All Itassociated partitions will be removeddeleted from all ISRs." }
    { "name": "Epoch", "type": "int64", "versions": "0+",
      "about": "The epoch of the broker to fenceas well." }
  ]
}

...

DelegationToken

Code Block
{
  "apiKey": 78,
  "type": "metadata",
  "name": "DeleteTopicRecordDelegationTokenRecord",
  "validVersions": "0",
  "fields": [
    { "name": "IdOwner", "type": "uuidstring", "versions": "0+",
      "about": "The topicdelegation totoken delete. All associated partitions will be deleted as well.owner." },
  ]
}

DelegationToken

Code Block
{
  "apiKey  { "name": 8"Renewers",
  "type": "metadata[]string",
  "nameversions": "DelegationTokenRecord0+",
      "validVersionsabout": "0",
  "fields": [The principals which have renewed this token." },
    { "name": "OwnerIssueTimestamp", "type": "stringint64", "versions": "0+",
      "about": "The delegation token owner "The time at which this timestamp was issued." },
    { "name": "RenewersMaxTimestamp", "type": "[]stringint64", "versions": "0+",
      "about": "The time principalsat which havethis token cannot be renewed thisany tokenmore." },
    { "name": "IssueTimestampExpirationTimestamp", "type": "int64", "versions": "0+",
      "about": "The next time at which this token timestampmust wasbe issuedrenewed." },
    { "name": "MaxTimestampTokenId", "type": "int64string", "versions": "0+",
      "about": "The time attoken id." },
  ]
}

ScramUser

Code Block
{
  "apiKey": 9,
  "type": "metadata",
  "name": "DelegationTokenRecord",
  "validVersions": "0",
  "fields": [ which this token cannot be renewed any more." },
    { "name": "ExpirationTimestampName", "type": "int64string", "versions": "0+",
      "about": "The next time at which this token must be reneweduser name." },
    { "name": "TokenIdCredentialInfos", "type": "string[]CredentialInfo", "versions": "0+",
      "about": "The token id." },
  ]
}

ScramUser

Code Block
{
  "apiKey": 9,
   mechanism and related information associated with the user's SCRAM credential.", "fields": [
      { "name": "Mechanism", "type": "metadataint8",
  "nameversions": "DelegationTokenRecord0+",
        "validVersionsabout": "0"The SCRAM mechanism." },
  "fields": [
    { "name": "NameSalt", "type": "stringbytes", "versions": "0+",
        "about": "The user nameA random salt generated by the client." },
      { "name": "CredentialInfosSaltedPassword", "type": "[]CredentialInfobytes", "versions": "0+",
        "about": "The mechanism and related information associated with the user's SCRAM credential.", "fields": [salted password." },
      { "name": "MechanismIterations", "type": "int8int32", "versions": "0+",
        "about": "TheThe number of iterations used in the SCRAM mechanismcredential." },]}
  ]
}

FeatureLevel

Code Block
{
   { "nameapiKey": "Salt"10,
  "type": "bytesmetadata",
  "versionsname": "0+FeatureLevelRecord",
        "aboutvalidVersions": "A0",
 random salt generated by the client." },
   "fields": [
    { "name": "SaltedPasswordName", "type": "bytesstring", "versions": "0+",
        "about": "The saltedfeature passwordname." },
      { "name": "IterationsFinalizedLevel", "type": "int32int16", "versions": "0+",
        "about": "The current finalized numberlevel of iterationsthis usedfeature infor the SCRAM credentialcluster." }]}
  ]
}

New Metrics

Full NameDescription

kafka.controller:type=KafkaController,name=MetadataLag

The offset delta between the latest metadata record this controller has replayed and the last stable offset of the metadata topic.

kafka.controller:type=KafkaServer,name=MetadataLagThe offset delta between the latest metadata record this broker has replayed and the last stable offset of the metadata topic.
kafka.controller:type=KafkaController,name=MetadataCommitLatencyMsThe latency of committing a message to the metadata topic.  Relevant on the active controller.
kafka.controller:type=KafkaController,name=MetadataCommitRateThe number of metadata messages per second committed to the metadata topic.
kafka.controller:type=KafkaController,name=MetadataSnapshotLagThe offset delta between the latest stable offset of the metadata topic and the offset of the last snapshot (or the last stable offset itself, if there are no snapshots)

...