Versions Compared

Old Version 18

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

Saved on

compared with

New Version 19

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

Saved on

Key

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

...

  • 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 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

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, group was empty for a configurable period, as for , the group metadata tombstone is written.

...

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

...

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." },
    { "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." },
    { "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
      "about": "The rebalance timeout." },

    { "name": "TopologyId", "type": "string", "versions": "0+",
      "about": "The ID of the topology. Must be non-empty." },

    { "name": "ProcessId", "type": "string", "versions": "0+",
      "about": "Identity of the streams instance that may have multiple consumers." },
    { "name": "UserEndpoint", "type": "stringEndpoint", "versions": "0+",
      "about": "User-defined endpoint for running interactive queries on this instance." },
    { "name": "ClientTags", "type": "[]KeyValue", "versions": "0+",
      "about": "Used for rack-aware assignment algorithm." }
  ],
  "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": "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" }
      ]
    }
  ]
}

...

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

...

  • 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

...

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

...

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

...

Code Block
languageyml
linenumberstrue
{
  "type": "data",
  "name": "StreamsGroupTopologyValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopologyId", "type": "string", "versions": "0+",
      "about": "The ID of the topology. Must be non-empty. " },
    { "name":  "Topology", "type": "[]Subtopology", "versions": "0+",
      "about": "The sub-topologies of the streams application.",
      "fields": [
        { "name": "Subtopology", "type": "string", "versions": "0+",
          "about": "String to uniquely identify the subtopology." },
        { "name": "SourceTopics", "type": "[]string", "versions": "0+",
          "about": "The topics the topology reads from." },
        { "name": "SourceTopicRegex", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
               "about": "The regular expressions identifying topics the topology reads from. null if not provided." },
        { "name": "SinkTopicsStateChangelogTopics", "type": "[]stringTopicInfo", "versions": "0+",
          "about": "The set of state changelog topics theassociated topologywith writesthis tosubtopology. " },
   	     { "name": "StateChangelogTopicsRepartitionSinkTopics", "type": "[]TopicInfostring", "versions": "0+",
          "about": "The set of state changelog repartition topics associatedthe withsub-topology thiswrites subtopologyto. " }, 
           { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
          "about": "The set of source topics that are internally created repartition topics. " }
      ]
    }
  ],
  "commonStructs": [
    { "name": "TopicConfig", "versions": "0+", "fields": [
      { "name": "key", "type": "string", "versions": "0+",
        "about": "The key of the topic-level configuration." },
      { "name": "value", "type": "string", "versions": "0+",
        "about": "The value of the topic-level configuration," }
    ]
    },
    { "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": "TopicConfigs", "type": "[]TopicConfig", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "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.

...

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

...

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

...

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

...

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

...