...

• A large fraction of the production incidents in Kafka Streams are caused by bugs in the rebalancing logic or misconfiguration thereof. With the rebalance logic being run on the processing node, the implementer of the business logic is also typically responsible for configuring, upgrading and debugging the rebalance logic. 

  • When rebalance problems occur, the timeline of events and information that caused the rebalance and influenced the assignment is spread across all clients, making it difficult to debug.

  • Since streams versions on the client are notoriously slow to be picked up by teams building the processing logic, bugs in the rebalance logic are prevalent for years after being fixed on trunk.

  • Since assignment logic is controlled by the streams application, tuning the parameters of the assignment logic requires redeployment of the application, which in itself triggers rebalances (and reassignments), creating a vicious cycle.

• With the rebalance logic being based on the metadata on one of the Streams clients, metadata inconsistencies among clients have caused a variety of issues.

• Particularly, the use of raw byte arrays in the protocol means that brokers cannot understand or validate the payload that is propagated among Streams clients.

• In the implementation, piggybacking the task assignment logic on the consumer partition assignment and injecting the required Streams metadata and functionality into the general-purpose KafkaConsumer interface has required a number of workarounds, increasing code complexity and hurting the maintainability of the code. This has affected both the streams code, which had to work around the restrictions of the consumer interface, and the consumer, whose interface had to be padded to account for Kafka Streams requirements.

• Because the protocol is not Kafka Streams native, advanced features like “smooth scale out” or “application shutdown on error” and others require workarounds like probing rebalances, which are disruptive for the running application as well as often hard to reason about by users.

...

Nothing particular.

Records

This section describes the new record types required for the new protocol. Similar to KIP-848, we define new record types that persist all data related to groups, their members and their assignments in the __consumer_offsets compacted topic. Each record has a dedicated key type that uses the version to differentiate the records from the records introduced in KIP-848, and those used to persist the committed offsets.

Compared to consumer groups, we introduce one extra record type, that stores information about the topology of the group.

The size of the persisted member metadata in the __consumer_offset topic will have to respect the max message size. But those records are persisted separately for each member, so it should be less of a concern. The same problem affects plain consumers in KIP-848 for very large consumer groups. Eventually, we may implement a way to split the records into multiple into multiple messages, and there are currently plans to add transaction support to the group coordinator to make sure the records can be updated atomically. For the first version of the protocol, we will limit those records by the max message size.

Group Metadata

StreamsGroupCurrentMemberAssignmentKey

The assignment-related metadata for a group with N members is stored with N+1 records. One metadata record per member and one metadata record for the group.

• 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 offsets and task lag offsets, which are not persisted, as they are constantly changing. 
• The group metadata, keyed by the group ID, contains the group epoch.

The records evolve in the following way:

• When the first member joins the group, the group metadata and the member metadata records for the member are first created.
• Every time a new member joins the group, another member metadata record is added
• Whenever a member updates its member metadata through the heartbeat, a new version of the member metadata record is appended.
• When a member leaves the group, a tombstone is written for the member metadata key.

When the last member leaves the group, the group metadata tombstone is written.

StreamsGroupMemberMetadataKey

{
  "type": "data",
 

{
  "type": "data",
  "name": "StreamsGroupCurrentMemberAssignmentKey",
  "validVersions": "14",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupIdStreamsGroupMemberMetadataKey",
  "typevalidVersions": "string11",
  "versionsflexibleVersions": "8none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "5",
      "about": "The group id." },
    { "name": "MemberId", "type": "string", "versions": "85",
      "about": "The member id." }
  ]
}

...

StreamsGroupMemberMetadataValue

{
  "type": "data",
  "name": "StreamsGroupCurrentMemberAssignmentValueStreamsGroupMemberMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MemberEpochInstanceId", "versions": "0+", "nullableVersions": "0+", "type": "int32string",
      "about": "The current(optional) member epoch that is expected from the member in the heartbeat requestinstance id." },
    { "name": "PreviousMemberEpochRackId", "versions": "0+", "nullableVersions": "0+", "type": "int32string",
      "about": "IfThe the(optional) last epoch bump is lost before reaching the member, the member will retry with the previous epochrack id." },
    { "name": "StateClientId", "versions": "0+", "type": "int8string",
      "about": "The memberclient stateid. See StreamsGroupMember.MemberState for the possible values." }" },
    { "name": "ActiveTasksClientHost", "versions": "0+", "type": "[]TaskIdsstring",
      "about": "CurrentlyThe assigned active tasks for this streams clientclient host." },
    { "name": "RebalanceTimeoutMs", "type": "StandbyTasksint32", "versions": "0+", "typedefault": "[]TaskIds"-1,
      "about": "CurrentlyThe assigned standby tasks for this streams clientrebalance timeout." },

    { "name": "WarmupTasksTopologyId", "versionstype": "0+string", "typeversions": "[]TaskIds0+",
      "about": "CurrentlyThe assignedID warm-upof tasksthe fortopology. thisMust streamsbe clientnon-empty." },

    { "name": "ActiveTasksPendingRevocationProcessId", "versionstype": "0+string", "typeversions": "[]TaskIds0+",
      "about": "TheIdentity activeof tasksthe thatstreams mustinstance bethat revokedmay byhave thismultiple memberconsumers." }
  ],
  "commonStructs": [
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyUserEndpoint", "type": "string", "versions": "0+",
        "about": "The subtopology identifierUser-defined endpoint for running interactive queries on this instance." },
      { "name": "PartitionsClientTags", "type": "[]int32KeyValue", "versions": "0+",
        "about": "TheUsed partitionsfor ofrack-aware the input topics processed by this member." }
    ]assignment algorithm." }
  ]
}

StreamsGroupMemberMetadataKey

{,
  "typecommonStructs": "data",[
    { "name": "StreamsGroupMemberMetadataKeyKeyValue",
  "validVersionsversions": "110+",
  "flexibleVersions": "none",
    "fields": [
        { "name": "GroupIdKey", "type": "string", "versions": "50+",
          "about": "Thekey group id.of the config" },
        { "name": "MemberIdValue", "type": "string", "versions": "50+",
          "about": "Thevalue of memberthe id."config" }
      ]
    }
  ]
}

...

StreamsGroupMetadataKey

{
  "type": "data",
  "name": "StreamsGroupMemberMetadataValueStreamsGroupMetadataKey",
  "validVersions": "09",
  "flexibleVersions": "0+none",
  "fields": [
    { "name": "InstanceIdGroupId", "versionstype": "0+string", "nullableVersionsversions": "0+", "type": "string3",
      "about": "The (optional)group instance id." },
  ]
}

StreamsGroupMetadataValue

{
  { "nametype": "RackIddata",
  "versionsname": "0+StreamsGroupMetadataValue",
  "nullableVersionsvalidVersions": "0+",
  "typeflexibleVersions": "string0+",
      "aboutfields": "The (optional) rack id." },[
    { "name": "ClientIdEpoch", "versions": "0+", "type": "stringint32",
      "about": "The clientgroup idepoch." }
  ]
}

Group Topology Metadata

The topology for the group, and the metadata for input topic partitions consumed in the topology, are persisted in two records.

• The partition metadata record, keyed by the group ID, contains metadata of all topic partitions consumed by the current topology
• The topology record, keyed by the group ID, stores the topology metadata sent as part of the StreamsGroupInitialize  request.

The records are evolved in the following way

• When the group coordinator receives a valid StreamsGroupInitialize  request, both records are created, using the topology sent in the request and the topic metadata currently known to the group coordinator.
• When the group coordinator detects that the partition metadata for one of the input partitions used in the topology changed (by comparing the previous partition metadata record to the currently known topic metadata in the broker) during the handling of a group heartbeat, the stream group partition metadata record is updated, and a new assignment is computed.

When the last member leaves the group, tombstones are written to remove the records for the group.

StreamsGroupPartitionMetadataKey

{
  "type": "data",
  "name": "StreamsGroupPartitionMetadataKey",
  "validVersions": "10",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "4",
      "about": "The group id." }
  ]
}

StreamsGroupPartitionMetadataValue

{
  "type": "data",
  "name": "StreamsGroupPartitionMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [,
    { "name": "ClientHost", "versions": "0+", "type": "string",
      "about": "The client host." },
    { "name": "RebalanceTimeoutMs", "type": "int32Topics", "versions": "0+", "defaulttype": -1"[]TopicMetadata",
      "about": "The list of rebalancetopic timeoutmetadata.", "fields": },[

      { "name": "TopologyIdTopicId", "typeversions": "string0+", "versionstype": "0+uuid",
        "about": "The ID of the topology. Must be non-emptytopic id." },

      { "name": "ProcessIdTopicName", "typeversions": "string0+", "versionstype": "0+string",
        "about": "Identity of the streams instance that may have multiple consumersThe topic name." },
      { "name": "UserEndpointNumPartitions", "typeversions": "string0+", "versionstype": "0+int32",
        "about": "User-definedThe endpointnumber forof runningpartitions interactiveof queries on this instancethe topic." },
      { "name": "ClientTagsPartitionMetadata", "typeversions": "[]KeyValue0+", "versionstype": "0+[]PartitionMetadata",
        "about": "UsedPartitions formapped rack-awareto assignmenta algorithm." }
  ],
  "commonStructs": [
    { "name": "KeyValue", "versions": "0+",
     set of racks. If the rack information is unavailable for all the partitions, an empty list is stored", "fields": [
          { "name": "KeyPartition", "typeversions": "string0+", "versionstype": "0+int32",
            "about": "keyThe of the configpartition number." },
          { "name": "ValueRacks", "typeversions": "string0+", "versionstype": "0+[]string",
            "about": "valueThe set of racks that the config partition is mapped to." }
      ]}
    ]}
  ]
}

...

StreamsGroupTopologyKey

{
  "type": "data",
  "name": "StreamsGroupMetadataKeyStreamsGroupTopologyKey",
  "validVersions": "915",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "3",
      "about": "The group id." }
  ]
}

...

StreamsGroupTopologyValue

{
  "type": "data",
  "name": "StreamsGroupMetadataValueStreamsGroupTopologyValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "EpochTopologyId", "versionstype": "0+string", "typeversions": "int320+",
      "about": "The ID groupof the epoch." }
  ]
}

StreamsGroupPartitionMetadataKey

{
  "type": "data",
 topology. Must be non-empty. " },
    { "name":  "StreamsGroupPartitionMetadataKeyTopology",
  "validVersionstype": "10[]Subtopology",
  "flexibleVersionsversions": "none0+",
  "fields": [
    { "nameabout": "GroupId", "type": "string", "versions": "4The sub-topologies of the streams application.",
      "about": "The group id." }
  ]
}

StreamsGroupPartitionMetadataValue

{
  "typefields": [
        { "name": "dataSubtopology",
  "nametype": "StreamsGroupPartitionMetadataValuestring",
  "validVersionsversions": "0+",
          "flexibleVersionsabout": "0+"String to uniquely identify the subtopology." },
  "fields": [
      { "name": "TopicsSourceTopics", "versionstype": "0+[]string", "typeversions": "[]TopicMetadata0+",
          "about": "The topics listthe oftopology topicreads metadatafrom." },
 "fields": [
      { "name": "SourceTopicRegex", "type": "TopicIdstring", "versions": "0+", "nullableVersions": "0+", "typedefault": "null",
           "about"uuid",
        "about": "The topic id: "The regular expressions identifying topics the topology reads from. null if not provided." },
        { "name": "TopicNameSinkTopics", "versionstype": "0+[]string", "typeversions": "string0+",
          "about": "The topic name topics the topology writes to." },
        { "name": "NumPartitionsStateChangelogTopics", "versionstype": "0+[]TopicInfo", "typeversions": "int320+",
          "about": "The numberset of partitions of the topic. state changelog topics associated with this subtopology. " },
        { "name": "PartitionMetadataRepartitionSourceTopics", "versionstype": "0+[]TopicInfo", "typeversions": "[]PartitionMetadata0+",
          "about": "Partitions mapped to a set of racks. If the rack information is unavailable for all the partitions, an empty list is storedThe set of source topics that are internally created repartition topics. " }
      ]
    }
  ],
  "commonStructs": [
    { "name": "TopicConfig", "versions": "0+", "fields": [
          { "name": "Partitionkey", "versionstype": "0+string", "typeversions": "int320+",
            "about": "The partition number key of the topic-level configuration." },
          { "name": "Racksvalue", "versionstype": "0+string", "typeversions": "[]string0+",
            "about": "The setvalue of racks that the partition is mapped to.topic-level configuration," }
      ]}
    ]},
  ]
}

StreamsGroupTargetAssignmentMemberKey

{
  "type  { "name": "TopicInfo", "versions": "0+", "fields": [
      { "name": "dataName",
  "nametype": "StreamsGroupTargetAssignmentMemberKeystring",
  "validVersionsversions": "130+",
        "flexibleVersionsabout": "none"The name of the topic." },
  "fields": [
    { "name": "GroupIdPartitions", "type": "stringint32", "versions": "70+",
        "about": "The group id number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      { "name": "MemberIdTopicConfigs", "type": "string[]TopicConfig", "versions": "0+", "nullableVersions": "0+", "default": "7null",
        "about": "The member id." Topic-level configurations as key-value pairs."
      }
    ]}
  ]
}

...

Current Member Assignment

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.

StreamsGroupCurrentMemberAssignmentKey

{
  "type": "data",
  "name": "StreamsGroupTargetAssignmentMemberValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ActiveTasks", "versionsname": "0+StreamsGroupCurrentMemberAssignmentKey",
  "typevalidVersions": "[]TaskIds14",
      "aboutflexibleVersions": "none"Currently,
 assigned active tasks for this streams client." }, "fields": [
    { "name": "StandbyTasksGroupId", "versionstype": "0+string", "typeversions": "[]TaskIds8",
      "about": "CurrentlyThe assigned standby tasks for this streams clientgroup id." },
    { "name": "WarmupTasksMemberId", "versionstype": "0+string", "typeversions": "[]TaskIds8",
      "about": "CurrentlyThe assigned warm-up tasks for this streams clientmember id." }
  ],
  "commonStructs": [
    { "name": 
}

StreamsGroupCurrentMemberAssignmentValue

{
  "type": "data",
 "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyStreamsGroupCurrentMemberAssignmentValue",
  "typevalidVersions": "string0",
  "versionsflexibleVersions": "0+",
        "aboutfields": "The subtopology identifier." },[
      { "name": "PartitionsMemberEpoch", "typeversions": "[]int320+", "versionstype": "0+int32",
        "about": "The partitionscurrent ofmember theepoch inputthat topicsis processedexpected byfrom thisthe member in the heartbeat request." },
    ]}
  ]
}

StreamsGroupTargetAssignmentMetadataKey

{
  "typename": "dataPreviousMemberEpoch",
  "nameversions": "StreamsGroupTargetAssignmentMetadataKey0+",
  "validVersionstype": "12int32",
      "flexibleVersionsabout": "none",
  "fields": [If the last epoch bump is lost before reaching the member, the member will retry with the previous epoch." },
    { "name": "GroupIdState", "typeversions": "string0+", "versionstype": "6int8",
      "about": "The group id member state. See StreamsGroupMember.MemberState for the possible values." },
  ]
}

StreamsGroupTargetAssignmentMetadataValue

{
  "type  { "name": "dataActiveTasks",
  "nameversions": "StreamsGroupTargetAssignmentMetadataValue0+",
  "validVersionstype": "0[]TaskIds",
      "flexibleVersionsabout": "0+",
  "fields": [Currently assigned active tasks for this streams client." },
    { "name": "AssignmentEpochStandbyTasks", "versions": "0+", "type": "int32[]TaskIds",
      "about": "The assignment epochCurrently assigned standby tasks for this streams client." },
   ]
}

StreamsGroupTopologyKey

{
  "type { "name": "dataWarmupTasks",
  "nameversions": "StreamsGroupTopologyKey0+",
  "validVersionstype": "15[]TaskIds",
      "flexibleVersionsabout": "none",
  "fields": [Currently assigned warm-up tasks for this streams client." },
    { "name": "GroupIdActiveTasksPendingRevocation", "typeversions": "string0+", "versionstype": "3[]TaskIds",
      "about": "The group id active tasks that must be revoked by this member." }
  ]
}

StreamsGroupTopologyValue

{,
  "typecommonStructs": "data",[
  "name": "StreamsGroupTopologyValue",
 { "validVersionsname": "0TaskIds",
  "flexibleVersionsversions": "0+",
  "fields": [
      { "name": "TopologyIdSubtopology", "type": "string", "versions": "0+",
        "about": "The ID of the topology. Must be non-empty. subtopology identifier." },
      { "name":  "TopologyPartitions", "type": "[]Subtopologyint32", "versions": "0+",
        "about": "The sub-topologies of the streams application.", partitions of the input topics processed by this member." }
      "fields": [
        { "name": "Subtopology", "type": "string", "versions": "0+",
          "about": "String to uniquely identify the subtopology." },
      ]}
  ]
}

Streams Group Target Assignment

The target assignment is stored in N + 1 records, where N is the number of members in the group. When a new target assignment is computed, the group coordinator will compare it with the current target assignment and only write the difference between the two assignments to the __consumer_offsets  partition. The assignment must be atomic, so the group coordinator will ensure that all the records are written in a single batch, limiting the size to the maximum batch size (default 1MB), as in consumer groups. When the last member leaves the group, a tombstone is written to remove the record.

StreamsGroupTargetAssignmentMemberKey

{
  "type": "data",
  "name": "StreamsGroupTargetAssignmentMemberKey",
  "validVersions": "13",
  "flexibleVersions": "none",
  "fields": [
      { "name": "SourceTopicsGroupId", "type": "[]string", "versions": "0+7",
          "about": "The topics the topology reads fromgroup id." },
        { "name": "SourceTopicRegexMemberId", "type": "string", "versions": "0+7",
      "nullableVersionsabout": "The member id." }
  ]
}

StreamsGroupTargetAssignmentMemberValue

{
  "type0+", "default": "nulldata",
           "aboutname": "The regular expressions identifying topics the topology reads from. null if not provided." } "StreamsGroupTargetAssignmentMemberValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
     { "name": "SinkTopicsActiveTasks", "typeversions": "[]string0+", "versionstype": "0+[]TaskIds",
          "about": "The topics the topology writes toCurrently assigned active tasks for this streams client." },
        { "name": "StateChangelogTopicsStandbyTasks", "typeversions": "[]TopicInfo0+", "versionstype": "0+[]TaskIds",
          "about": "TheCurrently setassigned ofstandby statetasks changelogfor topicsthis associated with this subtopology. streams client." },
        { "name": "RepartitionSourceTopics", "type": "[]TopicInfoWarmupTasks", "versions": "0+",
          "abouttype": "The set of source topics that are internally created repartition topics. " }
      ]
    "[]TaskIds",
      "about": "Currently assigned warm-up tasks for this streams client." }
  ],
  "commonStructs": [
    { "name": "TopicConfigTaskIds", "versions": "0+", "fields": [
      { "name": "keySubtopology", "type": "string", "versions": "0+",
        "about": "The key of the topic-level configurationsubtopology identifier." },
      { "name": "valuePartitions", "type": "string[]int32", "versions": "0+",
        "about": "The valuepartitions of the topic-level configuration," }
    ]input topics processed by this member." }
    ]},
    ]
}

StreamsGroupTargetAssignmentMetadataKey

{
  "type{ "name": "TopicInfo", "versions": "0+data", "fields": [
      { "name": "NameStreamsGroupTargetAssignmentMetadataKey",
  "typevalidVersions": "string12",
  "versionsflexibleVersions": "0+none",
        "about": "The name of the topic." },
  "fields": [
    { "name": "PartitionsGroupId", "type": "int32string", "versions": "0+6",
        "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      {group id." }
  ]
}

StreamsGroupTargetAssignmentMetadataValue

{
  "type": "data",
  "name": "StreamsGroupTargetAssignmentMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "AssignmentEpoch", "versions "name": "TopicConfigs", "type": "[]TopicConfig", "versions": "0+", "nullableVersions": "0+", "defaulttype": "nullint32",
        "about": "Topic-levelThe configurations as key-value pairs."
      }
    ]assignment epoch." }
  ]
}

Broker Metrics

The existing group metrics are extended to differentiate between streams groups and consumer groups and account for streams group states.

...

Another case is added to the org.apache.kafka.common.GroupType  enum:

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.

...