Versions Compared

Old Version 105

changes.mady.by.user Unknown User (lucasbru)

Saved on

compared with

New Version 106

changes.mady.by.user Unknown User (mjsax)

Saved on

Key

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

...

The member must set all the (top level) fields with the exception of RackId and InstanceId when it joins for the first time or when an error occurs (e.g. request timed out). Otherwise, it is expected to only fill in the fields which have changed since the last heartbeat.

Note

Update for Apache Kafka 4.3 release.

In Apache Kafka 4.2 release, there is a bug in the implementation of the StreamsGroupHeartbeatResponse using data type int32 for field AcceptableRecoveryLag instead of int64 as specified on the KIP. This bug has no direct impact on Kafka Streams 4.2 because the field is not used client side yet. To fix this bug, it is required to bump the request version of both StreamsGroupHeartbeatRequest and StreamsGroupHeartbestResponse to 1. If a Apache Kafka 4.3 broker receives an version-0 heartbeat, it will send back version-0 response and keeping acceptable-recovery-lag value unset using int32 type (as the client is older, expecting int32 and not using the field anyway). Only if the broker received as version-1 request, it will send back a version-1 response using correct int64 type for acceptable-recovery-lag field.


Code Block
languageyml
linenumberstrue
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupHeartbeatRequest",
  "validVersions": "0", // "0" in 4.2 release; bumped to "0-1" in 4.3 release
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
      "about": "The group identifier." },     
    { "name": "MemberId", "type": "string", "versions": "0+",
      "about": "The member ID generated by the streams consumer. The member ID must be kept during the entire lifetime of the streams consumer process." },
    { "name": "MemberEpoch", "type": "int32", "versions": "0+",
      "about": "The current member epoch; 0 to join the group; -1 to leave the group; -2 to indicate that the static member will rejoin." },
    { "name": "EndpointInformationEpoch", "type": "int32", "versions": "0+",
      "about": "The current endpoint epoch of this client, represents the latest endpoint epoch this client received"},
    { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "null if not provided or if it didn't change since the last heartbeat; the instance ID for static membership otherwise." },
    { "name": "RackId", "type": "string", "versions": "0+",  "nullableVersions": "0+", "default": "null",
      "about": "null if not provided or if it didn't change since the last heartbeat; the rack ID of the member otherwise." },
    { "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
      "about": "-1 if it didn't change since the last heartbeat; the maximum time in milliseconds that the coordinator will wait on the member to revoke its tasks otherwise." },

    { "name": "Topology", "type": "Topology", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The topology metadata of the streams application. Used to initialize the topology of the group and to check if the topology corresponds to the topology initialized for the group. Only sent when memberEpoch = 0, must be non-empty. Null otherwise.",
      "fields": [
        { "name": "Epoch", "type": "int32", "versions": "0+",
          "about": "The epoch of the topology. Used to check if the topology corresponds to the topology initialized on the brokers." },
        { "name": "Subtopologies", "type": "[]Subtopology", "versions": "0+",
          "about": "The sub-topologies of the streams application.",
          "fields": [
            { "name": "SubtopologyId", "type": "string", "versions": "0+",
              "about": "String to uniquely identify the subtopology. Deterministically generated from the topology" },
            { "name": "SourceTopics", "type": "[]string", "versions": "0+",
              "about": "The topics the topology reads from." },
            { "name": "SourceTopicRegex", "type": "[]string", "versions": "0+",
              "about": "The regular expressions identifying topics the subtopology reads from." },
            { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
              "about": "The set of state changelog topics associated with this subtopology. Created automatically." },
            { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
              "about": "The repartition topics the subtopology writes to." },
            { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
              "about": "The set of source topics that are internally created repartition topics. Created automatically." },
            { "name": "CopartitionGroups", "type": "[]CopartitionGroup", "versions": "0+",
              "about": "A subset of source topics that must be copartitioned.",
              "fields": [
                { "name": "SourceTopics", "type": "[]int16", "versions": "0+",
                  "about": "The topics the topology reads from. Index into the array on the subtopology level." },
                { "name": "SourceTopicRegex", "type": "[]int16", "versions": "0+",
                  "about": "Regular expressions identifying topics the subtopology reads from. Index into the array on the subtopology level." },
                { "name": "RepartitionSourceTopics", "type": "[]int16", "versions": "0+",
                  "about": "The set of source topics that are internally created repartition topics. Index into the array on the subtopology level." }
              ]}
          ]}
      ]
    },

    { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Currently owned active tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Currently owned standby tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Currently owned warm-up tasks for this client. Null if unchanged since last heartbeat." },

    { "name": "ProcessId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Identity of the streams instance that may have multiple consumers. Null if unchanged since last heartbeat." },
    { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "User-defined endpoint for Interactive Queries. Null if unchanged since last heartbeat or if not defined on the client." },
    { "name": "ClientTags", "type": "[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Used for rack-aware assignment algorithm. Null if unchanged since last heartbeat." },

    { "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Cumulative changelog offsets for tasks. Only updated when a warm-up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
    { "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Cumulative changelog end-offsets for tasks. Only updated when a warm-up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
    { "name": "ShutdownApplication", "type": "bool", "versions": "0+", "default": false,
      "about": "Whether all Streams clients in the group should shut down." }
  ],

  "commonStructs": [
    { "name": "KeyValue", "versions": "0+", "fields": [
      { "name": "Key", "type": "string", "versions": "0+",
        "about": "key of the config" },
      { "name": "Value", "type": "string", "versions": "0+",
        "about": "value of the config" }
    ]},
    { "name": "TopicInfo", "versions": "0+", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The name of the topic." },
      { "name": "Partitions", "type": "int32", "versions": "0+",
        "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      { "name": "ReplicationFactor", "type": "int16", "versions": "0+",
        "about": "The replication factor of the topic. Can be 0 if the default replication factor should be used." },
      { "name": "TopicConfigs", "type": "[]KeyValue", "versions": "0+",
        "about": "Topic-level configurations as key-value pairs."
      }
    ]},
    { "name": "Endpoint", "versions": "0+", "fields": [
      { "name": "Host", "type": "string", "versions": "0+",
        "about": "host of the endpoint" },
      { "name": "Port", "type": "uint16", "versions": "0+",
        "about": "port of the endpoint" }
    ]},
    { "name": "TaskOffset", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partition", "type": "int32", "versions": "0+",
        "about": "The partition." },
      { "name": "Offset", "type": "int64", "versions": "0+",
        "about": "The offset." }
    ]},
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]}
  ]
}

...

  1. Performs request validation.

  2. If the member joins the group (i.e. member epoch is 0):

    1. Look up or create the group.
      1. If the group is created the topology epoch of the group is set to the topology epoch sent by the member.
    2. GROUP_ID_NOT_FOUND is returned if the group ID is associated with a group type that is not streams or classic (the latter will be allowed for migration).
    3. Compare the topology epoch of the request ER to the topology epoch of the group EG
      1. If ER=EG, check if topology metadata in the request is equal to the topology metadata of the group, otherwise fail with STREAMS_INVALID_TOPOLOGY_EPOCH
      2. If ER<EG, fail with STREAMS_TOPOLOGY_FENCED 
      3. If ER>EG+1, fail with STREAMS_INVALID_TOPOLOGY_EPOCH
      4. If ER=EG+1, update the topology by writing the new topology record to the offset topic
    4. Creates the member.
  3. If the member is already part of the group (i.e. member epoch is greater than 0):
    1. Looks up the group.
    2. GROUP_ID_NOT_FOUND is returned if the group ID does not exist anymore.
    3. If the member does not exist, returns UNKNOWN_MEMBER_ID
    4. Checks whether the member epoch matches the member epoch in its current assignment. FENCED_MEMBER_EPOCH is returned otherwise. The member is also removed from the group.
      • There is an edge case here. When the group coordinator transitions a member to its target epoch, the heartbeat response with the new member epoch may be lost. In this case, the member will retry with the member epoch that it knows about and its request will be rejected with a FENCED_MEMBER_EPOCH. This will be handled as in KIP-848.
    5. If the topology epoch in the request is less than the topology epoch of the group, set state STALE_TOPOLOGY  in the response.
  4. Updates information of the member if needed. The group epoch is incremented if there is any change.
  5. If the topology or topic metadata changed, detect any topology / topic mismatches as described earlier in this document.
  6. Reconcile the member assignments as explained earlier in this document.
  7. When a member assignment gets changed it will bump the StreamsGroup EndpointEpoch  and bump the corresponding member EndpointInformationEpoch in the response.  When all members EndpointInformationEpoch values converge on the same value as the group-wide EndpointEpoch updated IQ information will be included in the member heartbeat response.  

Reponse Schema

...

  1. group-wide EndpointEpoch updated IQ information will be included in the member heartbeat response.  

Reponse Schema

The group coordinator will only set the ActiveTasks, StandbyTasks and WarmupTasks fields until the member acknowledges that it has converged to the desired assignment. This is done to ensure that the members converge to the target assignment.

Note

Update for Apache Kafka 4.3 release.

In Apache Kafka 4.2 release, there is a bug in the implementation of the StreamsGroupHeartbeatResponse using data type int32 for field AcceptableRecoveryLag instead of int64 as specified on the KIP. This bug has no direct impact on Kafka Streams 4.2 because the field is not used client side yet. To fix this bug, it is required to bump the request version of both StreamsGroupHeartbeatRequest and StreamsGroupHeartbestResponse to 1. If a Apache Kafka 4.3 broker receives an version-0 heartbeat, it will send back version-0 response and keeping acceptable-recovery-lag value unset using int32 type (as the client is older, expecting int32 and not using the field anyway). Only if the broker received as version-1 request, it will send back a version-1 response using correct int64 type for acceptable-recovery-lag field.


Code Block
languageyml
linenumberstrue
{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupHeartbeatResponse",
  "validVersions": "0",
   // "0" in 4.2 release; bumped to "0-1" in 4.3 release 
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - GROUP_ID_NOT_FOUND (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - UNKNOWN_MEMBER_ID (version 0+)
  // - FENCED_MEMBER_EPOCH (version 0+)
  // - UNRELEASED_INSTANCE_ID (version 0+)
  // - GROUP_MAX_SIZE_REACHED (version 0+)
  // - TOPIC_AUTHORIZATION_FAILED (version 0+) 
  // - CLUSTER_AUTHORIZATION_FAILED (version 0+)
  // - STREAMS_INVALID_TOPOLOGY (version 0+)
  // - STREAMS_INVALID_TOPOLOGY_EPOCH (version 0+)
  // - STREAMS_TOPOLOGY_FENCED (version 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 top-level 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": "MemberId", "type": "string", "versions": "0+",
      "about": "The member id is always generated by the streams consumer."},     
    { "name": "MemberEpoch", "type": "int32", "versions": "0+",
      "about": "The member epoch." },

    { "name": "HeartbeatIntervalMs", "type": "int32", "versions": "0+",
      "about": "The heartbeat interval in milliseconds." },
    { "name": "AcceptableRecoveryLag", "type": "int32", "versions": "0+",
      "about": "The maximal lag a warm-up task can have to be considered caught-up." },
     // incorrectly implemented as `int32` in 4.2 release; updated to `int64` and request/response version bump to "1" in 4.3 release.
    { "name": "TaskOffsetIntervalMs", "type": "int32", "versions": "0+",
      "about": "The interval in which the task changelog offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed." },

    { "name": "Status", "type":  "[]Status", "versions":  "0+",  "nullableVersions": "0+", "default": "null",
      "about": "Indicate zero or more status for the group.  Null if unchanged since last heartbeat." },

    // The streams app knows which partitions to fetch from given this information
    { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Assigned active tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Assigned standby tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Assigned warm-up tasks for this client. Null if unchanged since last heartbeat." },

    // IQ-related information
    { "name": "EndpointInformationEpoch", "type": "int32", "versions": "0+",
      "about": "The endpoint epoch set in the response"}, 
    { "name": "PartitionsByUserEndpoint", "type": "[]EndpointToPartitions", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Global assignment information used for IQ. Null if unchanged since last heartbeat." ,
      "fields": [
        { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+",
          "about": "User-defined endpoint to connect to the node" },
        { "name": "ActivePartitions", "type": "[]TopicPartition", "versions": "0+",
          "about": "All active partitions available on the node" },
        { "name": "StandbyPartitions", "type": "[]TopicPartition", "versions": "0+",
          "about": "All standby paritions available on the node"}
      ]
    }
  ],
  "commonStructs": [
    { "name": "Status", "versions": "0+", "fields": [
      // Possible status codes
      //  0 - STALE_TOPOLOGY                 - The topology epoch supplied is lower than the topology epoch for this streams group.
      //  1 - MISSING_SOURCE_TOPICS          - One or more source topics are missing or a source topic regex resolves to zero topics.
      //                                       Missing topics are indicated in the StatusDetail.
      //  2 - INCORRECTLY_PARTITIONED_TOPICS - One or more topics are incorrectly partitioned, that is, they are not copartitioned despite being 
      //                                       part of a copartition group, or the number of partitions in a changelog topic does not correspond
      //                                       to the maximal number of source topic partition for that subtopology.
      //                                       Incorrectly partitioned topics are indicated in the StatusDetail.
      //  3 - MISSING_INTERNAL_TOPICS        - One or more internal topics are missing. 
      //                                       Missing topics are indicated in the StatusDetail. 
      //                                       The group coordinator will attempt to create all missing internal topics, if any errors occur during
      //                                       topic creation, this will be indicated in StatusDetail.
      //  4 - SHUTDOWN_APPLICATION           - A client requested the shutdown of the whole application.
      //  5 - ASSIGNMENT_DELAYED             - No assignment was provided because assignment computation was delayed.
      { "name": "StatusCode", "type": "int8", "versions": "0+",
        "about": "A code to indicate that a particular status is active for the group membership" },
      { "name": "StatusDetail", "type": "string", "versions": "0+",
        "about": "A string representation of the status." }
    ]},
    { "name": "TopicPartition", "versions": "0+", "fields": [
      { "name": "Topic", "type": "string", "versions": "0+",
        "about": "topic name" },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "partitions" }
    ]},
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]},
    { "name": "Endpoint", "versions": "0+", "fields": [
      { "name": "Host", "type": "string", "versions": "0+",
        "about": "host of the endpoint" },
      { "name": "Port", "type": "uint16", "versions": "0+",
        "about": "port of the endpoint" }
    ]}
  ]
}

...