Page History

...

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. 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 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:

...

{
  "type": "data",
  "name": "StreamsGroupMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Epoch", "versions": "0+", "type": "int32",
      "about": "The group epoch." },
    { "name":  ]
}

Group Topology Metadata

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

"MetadataHash", "versions": "0+", "type": "int64",
      "about": "The hash of all topics in the group." },
    { "name": "ValidatedTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 0, "default": -1, "type": "int32",
      "about": "The topology epoch whose topics are validated to be present in a valid configuration in the metadata." },
    { "name": "LastAssignmentConfigs","taggedVersions": "0+", "nullableVersions": "0+","tag": 1, "default": null,
      "type": "[]LastAssignmentConfig", "about": "The last used configuration parameters as key-value pairs." }
  ],
  "commonStructs": [
    { "name": "LastAssignmentConfig", "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." }
    ]}
  ]
}

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 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 in the heartbeat request when members join.

...

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.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 tasks in all epochs since the assignment epoch of the task. 

StreamsGroupCurrentMemberAssignmentKey

...

{
  "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": "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": "SubtopologyId"{ "name": "AssignmentEpochs", "versions": "0+", "nullableVersions": "0+", "taggedVersions": "0+", "tag": 0, "type": "string[]int32", "versionsdefault": "0+"null,
        "about": "The epoch subtopologyat identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this memberwhich any 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

...

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.

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.

...