Versions Compared

Old Version 96

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

Saved on

compared with

New Version Current

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

Saved on

Key

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

...

  • Active tasks are reconciled like topic partitions in the KIP-848 consumer group protocol, that is, for reconciling the target assignment, the reconciliation follows three phases:

    • The group coordinator first asks the member to revoke any active tasks that are not assigned to that member in the target assignment any more (if either no warmup tasks are used at all, or if the target member has a hot standby/warmup task), by removing those active tasks from the set of tasks sent to the member in the heartbeat response

    • The member must first confirm revocation of these active tasks by removing them from the set of active tasks sent in the heartbeat request

    • Now, active tasks will be incrementally assigned to the member. An active task is assigned as soon as no other client owns it anymore, that is, once the previous owner (if any) has confirmed the revocation of the corresponding active task.

  • Standby and warm-up tasks are reconciled in parallel with active tasks, but following slightly different logic:

    • As for active tasks, standby and warm-up tasks removed from the members target assignment are removed from the assignment sent in the heartbeat response immediately, that is, with the next heartbeat. The member confirms the revocation of the tasks as soon as the state directory (or any other resource related to the task) is closed.

    • Newly assigned standby and warm-up tasks await that any member with the same processId owning that task (as active or standby) confirms revocation of the task by removing it from the corresponding set in a heartbeat response. That is, a task that is not an active task or a standby task on any other member with the same processId can be assigned immediately. Otherwise, it is assigned as soon as the blocking task is revoked. Target assignments that assign a task to two members with the same processId are invalid.

...

The possible states of the streams group are EMPTY, ASSIGNING, RECONCILING , STABLE, NOT_READY, DEAD. The states are similar as for consumer groups, but contain an extra stat NOT_READY, which is used when a group is not empty and has a topology set, but not all topics are present to start processing using the topology.

Global & Dynamic Group Configuration

We make core assignment options configurable centrally on the broker, without relying on each clients configuration. This allows tuning a streams group without redeploying the streams application. The three core assignment options will be introduced on the broker-side: acceptable.recovery.lag, num.warmup.replicas and , num.standby.replicas, and rack.aware.assignment.tags. They can be configured both globally on the broker, and dynamically for specific streams groups through the IncrementalAlterConfigs and DescribeConfigs RPCs. 

We store the last used assignment configuration in the group metadata on the broker. This way, we can detect if an assignment configuration is dynamically changed, and we can trigger reassignment immediately.

Online Migration from a Classic Consumer Group to a Streams Group

...

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.4 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/4.3 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.4 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
Code Block
languageyml
linenumberstrue
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupHeartbeatRequest",
  "validVersions": "0",
  "flexibleVersionslisteners": ["0+broker"],
  "fields": [
    { "name": "GroupIdStreamsGroupHeartbeatRequest",
 "type": "stringvalidVersions", "versions": "0+", "entityType":// "groupId0",
 in 4.2/4.3 release; bumped to "0-1" in 4.4 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 assignment epochendpoint 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." }
    ]}
  ]
}

...

  • READ on group

  • CREATE on cluster resource, or CREATE on all topics in StateChangelogTopics and RepartitionSourceTopics

    • Note that this ACLs are only required if the group coordinator should create the internal topics implicitly. If the internal topics are created explicitly, this ACL is not needed for the Streams group heartbeat.

  • DESCRIBE_CONFIGS on  on all topics included in the message

...

  • GroupId must be non-empty.

  • MemberId must be non-empty.

  • MemberEpoch must be >= -2.

  • InstanceId, if not null, must be non-empty.

  • RebalanceTimeoutMs must be larger than zero in the first heartbeat request.

  • ActiveTasks, StandbyTasks and WarmupTasks have to be disjoint sets

  • ActiveTasks, StandbyTasks and WarmupTasks have to be non-null and empty when joining (member epoch is 0)

  • If MemberEpoch  is 0, Topology should be non-null. Otherwise, it should be null.
  • If EndpointInformationEpoch is 0, Topology should be non-null. Otherwise, it should be null.

STREAMS_INVALID_TOPOLOGY is returned when the request contains a new topology and should the topology not obey the following invariants:

...

  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.  

...

Response 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.4 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/4.3 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.4 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
Code Block
languageyml
linenumberstrue
{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupHeartbeatResponse",
  "validVersionsname": "0StreamsGroupHeartbeatResponse",
  "flexibleVersionsvalidVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)"0" in 4.2/4.3 release; bumped to "0-1" in 4.4 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.4 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"},
      ]
     {}
  ],
  "namecommonStructs": "EndpointInformationEpoch",[
    { "typename": "int32Status", "versions": "0+",
          "about": "The currentAssignment epoch"}        
      ]
    }
  ],
  "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.
      { "name": "StatusCode", "type": "int8", "versions": "0+",
//  5 - ASSIGNMENT_DELAYED          "about": "A code to- indicateNo thatassignment awas particularprovided statusbecause isassignment activecomputation forwas thedelayed.
 group membership" },
   //  6 { "name": "StatusDetail", "type": "string", "versions": "0+",
- MISSING_CLIENT_TAGS             "about": "A string - (Response v1+ only) The GC is configured to use `client.tag.` for rack-aware standby task assignment, but the client does not provide this tag
      { "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" }
    ]}
  ]
}

...

Code Block
languageyml
linenumberstrue
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupDescribeRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupIds", "type": "[]string", "versions": "0+", "entityType": "groupId",
      "about": "The ids of the groups to describe" },
    { "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+",
      "about": "Whether to include authorized operations." }
  ]
}

Required ACL

  • Read Group

Request Validation

  • DESCRIBE on the group resource

  • DESCRIBE on all topics used in the group's topology

Request Validation

INVALID_REQUEST INVALID_REQUEST is returned should the request not obey the schema definition.

...

  • Checks whether the group ids exists. If it does not, GROUP_ID_NOT_FOUND is returned. Similarly, it is returned if the group ID is associated with a group type that is not streams .

  • Looks up the groups and returns the response.

Response Schema


Code Block
languageyml
linenumberstrue
{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupDescribeResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - INVALID_GROUP_ID (version 0+)
  // - GROUP_ID_NOT_FOUND (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": "Groups", "type": "[]DescribedGroup", "versions": "0+",
      "about": "Each described group.",
      "fields": [
        { "name": "ErrorCode", "type": "int16", "versions": "0+",
          "about": "The describe error, 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": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
          "about": "The group ID string." },
        { "name": "GroupState", "type": "string", "versions": "0+",
          "about": "The group state string, or the empty string." },
        { "name": "GroupEpoch", "type": "int32", "versions": "0+",
          "about": "The group epoch." },
        { "name": "AssignmentEpoch", "type": "int32", "versions": "0+",
          "about": "The assignment epoch." },

        { "name":  "Topology", "type": "Topology", "versions": "0+", "nullableVersions": "0+", "default": "null",
          "about": "The topology metadata currently initialized for the streams application. Can be null in case of a describe error.",
          "fields": [
            { "name": "Epoch", "type": "int32", "versions": "0+",
              "about": "The epoch of the currently initialized topology for this group." },
            { "name": "Subtopologies", "type": "[]Subtopology", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The subtopologies of the streams application. This contains the configured subtopologies, where the number of partitions are set and any regular expressions are resolved to actual topics. Null if the group is uninitialized, source topics are missing or incorrectly partitioned.",
              "fields": [
                { "name": "SubtopologyId", "type": "string", "versions": "0+",
                  "about": "String to uniquely identify the subtopology." },
                { "name": "SourceTopics", "type": "[]string", "versions": "0+",
                  "about": "The topics the subtopology reads from." },
                { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
                  "about": "The repartition topics the subtopology writes to." },
                { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
                  "about": "The set of state changelog topics associated with this subtopology. Created automatically." },
                { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
                  "about": "The set of source topics that are internally created repartition topics. Created automatically." }
              ]}
          ]},
        { "name": "Members", "type": "[]Member", "versions": "0+",
          "about": "The members.",
          "fields": [
            { "name": "MemberId", "type": "string", "versions": "0+",
              "about": "The member ID." },
            { "name": "MemberEpoch", "type": "int32", "versions": "0+",
              "about": "The member epoch." },
  
            { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member instance ID for static membership." },
            { "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The rack ID." },

            { "name": "ClientId", "type": "string", "versions": "0+",
              "about": "The client ID." },
            { "name": "ClientHost", "type": "string", "versions": "0+",
              "about": "The client host." },

            { "name": "TopologyEpoch", "type": "int32", "versions": "0+",
              "about": "The epoch of the topology on the client." },

            { "name": "ProcessId", "type": "string", "versions": "0+",
              "about": "Identity of the streams instance that may have multiple clients. " },
            { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "User-defined endpoint for Interactive Queries. Null if not defined for this client." },
            { "name": "ClientTags", "type": "[]KeyValue", "versions": "0+",
              "about": "Used for rack-aware assignment algorithm." },
            { "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+",
              "about": "Cumulative changelog offsets for tasks." },
            { "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+",
              "about": "Cumulative changelog end offsets for tasks." },

            { "name": "Assignment", "type": "Assignment", "versions": "0+",
              "about": "The current assignment." },
            { "name": "TargetAssignment", "type": "Assignment", "versions": "0+",
              "about": "The target assignment." },
            { "name": "IsClassic", "type": "bool", "versions": "0+",
              "about": "True for classic members that have not been upgraded yet." }
        ]},
        { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
          "about": "32-bit bitfield to represent authorized operations for this group." }
      ]
    }
  ],
  "commonStructs": [
    { "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": "TopicPartitions", "versions": "0+", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic ID." },
      { "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions." }
    ]},
    { "name": "Assignment", "versions": "0+", "fields": [
      { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Active tasks for this client." },
      { "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Standby tasks for this client." },
      { "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Warm-up tasks for this client. " }
    ]},
    { "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": "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."
      }
    ]}
  ]
}

...

  • The member metadata record, keyed by the group ID and the member ID, stores all relevant metadata that is sent by a member to the coordinator in a heartbeat request. A notable exception are task changelog offsets and task changelog end offsets, which are not persisted, as they are constantly changing. 
  • The group metadata, keyed by the group ID, contains the group epoch.
    • The group epoch
    • A hash of the topology metadata
    • The epoch of the last validated topology
    • The last configuration used by the assignor

The records evolve in the following way:

...

Code Block
languageyml
linenumberstrue
{
  "type": "data",
  "name": "StreamsGroupMemberMetadataKey",
  "validVersions": TBD,
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": TBD,
      "about": "The group id." },
    { "name": "MemberId", "type": "string", "versions": TBD,
      "about": "The member id." }
  ]
}

StreamsGroupMemberMetadataValue


Code Block
languageyml
linenumberstrue
{
  "type": "data",
  "name": "StreamsGroupMemberMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "InstanceId", "versions": "0+", "nullableVersions": "0+", "type": "string",
      "about": "The (optional) instance ID for static membership." },
    { "name": "RackId", "versions": "0+", "nullableVersions": "0+", "type": "string",
      "about": "The (optional) rack id." },
    { "name": "ClientId", "versions": "0+", "type": "string",
      "about": "The client id." },
    { "name": "ClientHost", "versions": "0+", "type": "string",
      "about": "The client host." },
    { {
  "type": "data",
  "name": "StreamsGroupMemberMetadataKeyRebalanceTimeoutMs",
  "validVersionstype": TBD"int32",
 "versions": "flexibleVersions0+":, "nonedefault": -1,
      "fieldsabout": [ "The rebalance timeout." },

    { "name": "GroupIdTopologyEpoch", "type": "stringint32", "versions": TBD"0+",
      "about": "The group id epoch of the topology. Must be non-zero." },

    { "name": "MemberIdProcessId", "type": "string", "versions": TBD"0+",
      "about": "The member idIdentity of the streams instance that may have multiple consumers." },
  ]
}

StreamsGroupMemberMetadataValue

Code Block
languageyml
linenumberstrue
{
  "type  { "name": "dataUserEndpoint",
  "nametype": "StreamsGroupMemberMetadataValueEndpoint",
  "validVersionsversions": "0+",
  "flexibleVersionsnullableVersions": "0+",
  "fieldsdefault": ["null",
      { "nameabout": "InstanceId", "versionsUser-defined endpoint for running interactive queries on this instance." },
    { "name": "0+ClientTags", "nullableVersionstype": "0+[]KeyValue", "typeversions": "string0+",
      "about": "TheUsed (optional) instance ID for static membershipfor rack-aware assignment algorithm." }
  ],
  "commonStructs": [
    { "name": "RackIdEndpoint", "versions": "0+", "nullableVersionsfields": "0+", "type": "string",
[
      { "aboutname": "The (optional) rack id." },
    { "nameHost", "type": "ClientIdstring", "versions": "0+",
 "type": "string",
      "about": "Thehost of clientthe id.endpoint" },
      { "name": "ClientHostPort", "versionstype": "0+uint16", "typeversions": "string0+",
        "about": "Theport of clientthe host.endpoint" }
    ]},
    { "name": "RebalanceTimeoutMs", "type": "int32"KeyValue", "versions": "0+", "default": -1,
      "aboutfields": "The[
  rebalance timeout." },

    { "name": "TopologyEpochKey", "type": "int32string", "versions": "0+",
          "about": "Thekey epoch of the topology. Must be non-zero. config" },

        { "name": "ProcessIdValue", "type": "string", "versions": "0+",
          "about": "Identityvalue of the streamsconfig" instance}
 that may have multiple consumers." },]
    }
  ]
}

StreamsGroupMetadataKey


Code Block
languageyml
linenumberstrue
{
 "name": "UserEndpoint", "type": "Endpointdata",
  "versionsname": "0+StreamsGroupMetadataKey",
  "nullableVersionsvalidVersions": "0+",TBD,
  "defaultflexibleVersions": "nullnone",
      "aboutfields": "User-defined endpoint for running interactive queries on this instance." },[
    { "name": "ClientTagsGroupId", "type": "[]KeyValuestring", "versions": "0+"TBD,
      "about": "UsedThe for rack-aware assignment algorithmgroup id." }
  ]
}
 

StreamsGroupMetadataValue


Code Block
languageyml
linenumberstrue
{,
  "commonStructstype": ["data",
  "name": "StreamsGroupMetadataValue",
 { "namevalidVersions": "Endpoint0",
  "versionsflexibleVersions": "0+",
  "fields": [
      { "name": "HostEpoch", "typeversions": "string0+", "versionstype": "0+int32",
        "about": "host of"The thegroup endpointepoch." },
      { "name": "PortMetadataHash", "typeversions": "uint160+", "versionstype": "0+int64",
        "about": "portThe hash of theall endpoint"topics }
in the  group." ]},
    { "name": "KeyValueValidatedTopologyEpoch", "versions": "0+",
      "fields": [
        { "name": "Key""taggedVersions": "0+", "tag": 0, "default": -1, "type": "stringint32",
      "versionsabout": "0+",
The topology epoch whose topics are validated to be  "about": "key ofpresent in a valid configuration in the configmetadata." },
        { "name": "ValueLastAssignmentConfigs", "typetaggedVersions": "string0+", "versionsnullableVersions": "0+",
   "tag": 1, "default": null,
       "abouttype": "[]LastAssignmentConfig"value of the config" }
      ]
    }
  ]
}

StreamsGroupMetadataKey

Code Block
languageyml
linenumberstrue
{
  "type": "data", "about": "The last used configuration parameters as key-value pairs." }
  ],
  "namecommonStructs": "StreamsGroupMetadataKey",[
    { "validVersionsname": TBD"LastAssignmentConfig",
  "flexibleVersionsversions": "none0+",
  "fields": [
      { "name": "GroupIdKey", "type": "string", "versions": TBD"0+",
        "about": "TheKey of groupthe idconfig." },
  ]
}

StreamsGroupMetadataValue

Code Block
languageyml
linenumberstrue
{
  "type": "data",
 { "name": "StreamsGroupMetadataValueValue",
  "validVersionstype": "0string",
  "flexibleVersionsversions": "0+",
  "fields": [
    { "nameabout": "Epoch", "versions": "0+", "type": "int32",Value of the config." }
      "about": "The group epoch." ]}
  ]
}

Group Topology Metadata

...

For each member, there is a record to store the current member assignment. The record is first created when a member joins (with an empty assignment). Every time an assignment is computed for the member during reconciliation, it is compared to the previous assignment for the member. If the assignment changed, a new version of the current member assignment record is appended to the consumer offset topic. When the member leaves the group, a tombstone is written to remove the record.

Together with the assignment, we track the epoch at which an active task was assigned. This helps with offset commit validation: We allow offset commits for partitions that belong to the consumer offset topic. When the member leaves the group, a tombstone is written to remove the record.tasks in all epochs since the assignment epoch of the task. 

StreamsGroupCurrentMemberAssignmentKey

...

Code Block
languageyml
linenumberstrue
{
  "type": "data",
  "name": "StreamsGroupCurrentMemberAssignmentValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MemberEpoch", "versions": "0+", "type": "int32",
      "about": "The current member epoch that is expected from the member in the heartbeat request." },
    { "name": "PreviousMemberEpoch", "versions": "0+", "type": "int32",
      "about": "If the last epoch bump is lost before reaching the member, the member will retry with the previous epoch." },
    { "name": "State", "versions": "0+", "type": "int8",
      "about": "The member state. See StreamsGroupMember.MemberState for the possible values." },
    { "name": "ActiveTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned active tasks for this streams client." },
    { "name": "StandbyTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned standby tasks for this streams client." },
    { "name": "WarmupTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned warm-up tasks for this streams client." },
    { "name": "ActiveTasksPendingRevocation", "versions": "0+", "type": "[]TaskIds",
      "about": "The active tasks that must be revoked by this member." },
    { "name": "StandbyTasksPendingRevocation", "versions": "0+", "type": "[]TaskIds",
      "about": "The standby tasks that must be revoked by this member." },
    { "name": "WarmupTasksPendingRevocation", "versions": "0+", "type": "[]TaskIds",
      "about": "The warmup tasks that must be revoked by this member." }
  ],
  "commonStructs": [
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type":  { "name": "SubtopologyId""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": "AssignmentEpochs", "versions": "0+", "nullableVersions": "0+", "taggedVersions": "0+", "tag": 0, "type": "string[]int32", "versionsdefault": "0+"null,
        "about": "The epoch at subtopologywhich identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this memberany partition was assigned to the member. Used to fence zombie commits requests. Of the same length as partitions. If null, all assignment epochs are considered to be equal to the member epoch." }
      ]}
  ]
}

Streams Group Target Assignment

...

Name

Type

Default

Doc

group.streams.session.timeout.ms

int

45s

The timeout to detect client failures when using the streams group protocol.

group.streams.min.session.timeout.ms

int

45s

The minimum session timeout.

group.streams.max.session.timeout.ms

int

60s

The maximum session timeout.

group.streams.heartbeat.interval.ms

int

5s

The heartbeat interval given to the members.

group.streams.min.heartbeat.interval.ms

int

5s

The minimum heartbeat interval.

group.streams.max.heartbeat.interval.ms

int

15s

The maximum heartbeat interval.

group.streams.max.size

int

MaxValue

The maximum number of streams clients that a single streams group can accommodate.

group.streams.acceptable.recovery.lag

long

10’000

The maximum acceptable lag (number of offsets to catch up) for a client to be considered caught-up enough to receive an active task assignment.

group.streams.num.warmup.replicas

int

2

The maximum number of warmup replicas.

group.streams.max.warmup.replicas

int

20

Maximum for dynamic configurations of the warmup replica configuration

group.streams.num.standby.replicas

int

0

The number of standby replicas for each task.

group.streams.max.standby.replicas

int

2

Maximum for dynamic configurations of the standby replica configuration

group.streams.task.offset.interval.ms

int

60s

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.

group.streams.min.task.offset.interval.ms

int

15s

Minimum for dynamic configurations of the task offset interval. Used to restrict users of the cluster from sending the task changelog offsets too often.

group.streams.assignor.name

string

null

The name of the task assignor used for all streams groups. Can be sticky or highly_available in AK.

group.streams.initial.rebalance.delay.ms

int

3000

The first rebalance of a group is delayed by this amount to allow more members to join the group.

group.streams.rack.aware.assignment.tags

string[]

empty list

The client tags used for rack aware standby task assignment.

Group Configurations

We will add new configurations for the resource type GROUP in DescribeConfigs and IncrementalAlterConfigs to override the default broker configurations dynamically for specific groups.

Name

Type

DefaultDoc

group.streams.session.timeout.ms

int

45sThe timeout to detect client failures when using the streams group protocol.

group.streams.heartbeat.interval.ms

int

5s

The heartbeat interval given to the members.

group.streams.acceptable.recovery.lag

long

10’000The maximum acceptable lag (number of offsets to catch up) for a client to be considered caught-up enough to receive an active task assignment.

group.streams.num.warmup.replicas

int

2

The maximum number of warmup replicas.

group.streams.num.standby.replicas

int

0The number of standby replicas for each task.

group.streams.task.offset.interval.ms

int

60s

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.

group.streams.assignor.name

string

null

The name of the task assignor used for this streams groups. Can be sticky or highly_available in AK.

streams.initial.rebalance.delay.ms

int

3000

streams.rack.aware.assignment.tags

string[]

empty list

Streams API

In a future major version, when the classic group protocol will be deprecated, the interfaces introduced in KIP-924 would be deprecated, unless an integration with this KIP is proposed in a follow-up KIP.

...

Option

Description

--version

Display Kafka version.

--verbose

Use with --describe --state  to show group epoch and target assignment epoch.

Use with --describe --members  to show for each member the member epoch, target assignment epoch, current assignment, target assignment, and whether member is still using the classic rebalance protocol (KIP-1099).

Use with --describe --offsets  and --describe  to show leader epochs (KIP-320).

--all-input-topics

Use with --reset-offsets or --delete-offsets. If specified, includes all input topics of the streams group, as stored by the topology metadata on the broker. 

--input-topics topic <String: topics>topic>

Use with --reset-offsets or --delete-offsets. Comma-separated list of user input Defines a user input topic to be modified. Multiple such options can be used to define multiple topics. For these topics, the tool by default will reset the offset to the earliest available offset, or delete the offsets. Reset to other offset position by appending other reset offset option, ex: --input-topics topic foo --shift-by 5. 

--internal-topics <String: topics>

Use with --delete. Comma-separated list of internal topics to delete. Must be a subset of the internal topics marked for deletion by the default behaviour (do a dry-run without this option to view these topics). 

--to-offset <Long: offset>

Reset input topic offsets to a specific offset

--to-latest

Reset input topic offsets to latest offset.

--to-earliest

Reset input topic offsets to earliest offset.

--by-duration <String: duration>

Reset input topic offsets to offset by duration from current timestamp. Format: 'PnDTnHnMnS'

--to-datetime <String: datetime>

Reset input topic offsets to offset from datetime. Format: 'YYYY-MM-DDTHH:mm:SS.sss'.

--from-file

Reset input topic offsets to values defined in CSV file.

--shift-by <Long: n>

Reset input topic offsets shifting current offset by 'n', where 'n' can be positive or negative.

--timeout <Long: timeout (ms)>

The timeout that can be set for some use cases. For example, it can be used when describing the group to specify the maximum amount of time in milliseconds to wait before the group stabilizes (when the group is just created, or is going through some changes). (default: 5000)   

--state [String]

When specified with '--describe', includes the state of the group. When specified with '--list', it displays the state of all groups. It can also be used to list groups with specific states. The valid values are 'Empty', 'Not Ready', 'Reconciling', 'Assigning', 'Stable' and 'Dead'.

--reset-offsets

Reset input topic offsets of streams group. Supports one streams group at a time, and instances should be inactive.

You must choose one of the following reset specifications: --to-datetime, --by-duration, --to-earliest, --to-latest, --shift-by, --from-file, --to-current, --to-offset.

To define the scope use --all-input-topics or --input-topics. One scope must be specified unless you use '--from-file'.

Fails if neither '--dry-run' nor '–execute--execute' is specified.

--offsets

Describe the group and list all input topic partitions in the group along with their offset lag. This is the default sub-action of --describe and may be used with the '--describe' option only.

--members

Describe members of the group. This option may be used with the '--describe' option only.

--list

List all streams groups.

--help

Print usage information.

--group <String: streams group ID>

The group ID (application ID) of the streams ID we wish to act on.

--all-groups

To act on all groups. Only allowed with --delete

--execute

Execute operation. Supported operations: reset-offsets.

--dry-run

Only show results without executing changes on streams groups. Supported operations: reset-offsets.

--describe

Describe streams group and list offset lag (number of records not yet processed) related to given group.

--delete-offsets

Delete offsets of streams group. Supports one streams group at the time.

To define the scope use --all-input-topics or --input-topics. One scope must be specified unless you use '--from-file'.

--delete

Pass in a group to delete entire streams group. For instance --group g1. Deletes offsets, internal topics of the streams group and and topology and ownership information on the broker. Instances should be inactive.

--command-config <String: command config property file>

Property file containing configs to be passed to Admin Client.

--bootstrap-server <String: server to connect to>

REQUIRED: The server(s) to connect to.

...