Copilot commented on code in PR #21170: URL: https://github.com/apache/kafka/pull/21170#discussion_r2630351868
########## docs/streams/developer-guide/streams-rebalance-protocol.html: ########## @@ -0,0 +1,351 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +<script><!--#include virtual="../../js/templateData.js" --></script> + +<script id="content-template" type="text/x-handlebars-template"> + <!-- h1>Developer Guide for Kafka Streams</h1 --> + <div class="sub-nav-sticky"> + <div class="sticky-top"> + <!-- div style="height:35px"> + <a href="/{{version}}/documentation/streams/">Introduction</a> + <a class="active-menu-item" href="/{{version}}/documentation/streams/developer-guide">Developer Guide</a> + <a href="/{{version}}/documentation/streams/core-concepts">Concepts</a> + <a href="/{{version}}/documentation/streams/quickstart">Run Demo App</a> + <a href="/{{version}}/documentation/streams/tutorial">Tutorial: Write App</a> + </div --> + </div> + </div> + + <div class="section" id="streams-rebalance-protocol"> + <h1>Streams Rebalance Protocol<a class="headerlink" href="#streams-rebalance-protocol" title="Permalink to this headline"></a></h1> + <p>The Streams Rebalance Protocol is a broker-driven rebalancing system designed specifically for Kafka Streams applications. Following the pattern of KIP-848, which moved rebalance coordination of plain consumers from clients to brokers, KIP-1071 extends this model to Kafka Streams workloads.</p> + + <div class="contents local topic" id="table-of-contents"> + <p class="topic-title first"><b>Table of Contents</b></p> + <ul class="simple"> + <li><a class="reference internal" href="#overview" id="id1">Overview</a></li> + <li><a class="reference internal" href="#whats-supported" id="id2">What's Supported in This Version</a></li> + <li><a class="reference internal" href="#whats-not-supported" id="id3">What's Not Supported in This Version</a></li> + <li><a class="reference internal" href="#why-use" id="id4">Why Use the Streams Rebalance Protocol?</a></li> + <li><a class="reference internal" href="#enabling" id="id5">Enabling the Protocol</a></li> + <li><a class="reference internal" href="#configuration" id="id6">Configuration</a><ul> + <li><a class="reference internal" href="#broker-configuration" id="id7">Broker Configuration</a></li> + <li><a class="reference internal" href="#group-configuration" id="id8">Group Configuration</a></li> + <li><a class="reference internal" href="#streams-configuration" id="id9">Streams Configuration</a></li> + </ul> + </li> + <li><a class="reference internal" href="#administration" id="id10">Administration</a></li> + <li><a class="reference internal" href="#architecture" id="id11">Architecture and How It Works</a><ul> + <li><a class="reference internal" href="#streams-groups" id="id12">Streams Groups</a></li> + <li><a class="reference internal" href="#topology-configuration" id="id13">Topology Configuration and Validation</a></li> + <li><a class="reference internal" href="#assignment-configuration" id="id14">Centralized Assignment Configuration</a></li> + </ul> + </li> + <li><a class="reference internal" href="#monitoring" id="id15">Monitoring and Metrics</a></li> + <li><a class="reference internal" href="#migration" id="id16">Migration from Classic Protocol</a></li> + </ul> + </div> + + <div class="section" id="overview"> + <h2><a class="toc-backref" href="#id1">Overview</a><a class="headerlink" href="#overview" title="Permalink to this headline"></a></h2> + <p>Instead of clients computing new assignments on the client during rebalance events involving all members of the group, assignments are computed continuously on the broker. Instead of using a consumer group, the streams application registers as a <strong>streams group</strong> with the broker, which manages and exposes all metadata required for coordination of the streams application instances.</p> + + <p>This approach brings Kafka Streams coordination in line with the modern broker-driven rebalance model introduced for consumers in KIP-848, providing a dedicated group type with streams-specific semantics and metadata management.</p> + </div> + + <div class="section" id="whats-supported"> + <h2><a class="toc-backref" href="#id2">What's Supported in This Version</a><a class="headerlink" href="#whats-supported" title="Permalink to this headline"></a></h2> + <p>The following features are available in the current release:</p> + <ul class="simple"> + <li><strong>Core Streams Group Rebalance Protocol</strong>: The <code>group.protocol=streams</code> configuration enables the dedicated streams rebalance protocol. This separates streams groups from consumer groups and provides a streams-specific group membership lifecycle and metadata management on the broker.</li> + + <li><strong>Sticky Task Assignor</strong>: A basic task assignment strategy that minimizes task movement during rebalances is included.</li> + + <li><strong>Interactive Query Support</strong>: IQ operations are compatible with the new streams protocol.</li> + + <li><strong>New Admin RPC</strong>: The StreamsGroupDescribe RPC provides streams-specific metadata separate from consumer group information, with corresponding access via the <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/admin/Admin.html">Admin client</a>.</li> + + <li><strong>CLI Integration</strong>: You can list, describe, and delete streams groups via the <a class="reference internal" href="kafka-streams-group-sh.html">kafka-streams-groups.sh</a> script.</li> + + <li><strong>Offline Migration</strong>: After shutting down all members and waiting for their <code>session.timeout.ms</code> to expire, a classic group can be converted to a streams group and a streams group can be converted to a classic group. The only thing that will be preserved broker-side are the committed offsets of the application.</li> + </ul> + </div> + + <div class="section" id="whats-not-supported"> + <h2><a class="toc-backref" href="#id3">What's Not Supported in This Version</a><a class="headerlink" href="#whats-not-supported" title="Permalink to this headline"></a></h2> + <p>The following features are not yet available and should be avoided when using the new protocol:</p> + <ul class="simple"> + <li><strong>Static Membership</strong>: Setting a client <code>instance.id</code> will be rejected.</li> + + <li><strong>Topology Updates</strong>: If a topology is changed significantly (e.g., by adding new source topics or changing the number of sub-topologies), a new streams group must be created.</li> + + <li><strong>High Availability Assignor</strong>: Only the sticky assignor is supported.</li> + + <li><strong>Regular Expressions</strong>: Pattern-based topic subscription is not supported.</li> + + <li><strong>Online Migration</strong>: Group migration while the application is running is not available between the classic and new streams protocol.</li> + </ul> + </div> + + <div class="section" id="why-use"> + <h2><a class="toc-backref" href="#id4">Why Use the Streams Rebalance Protocol?</a><a class="headerlink" href="#why-use" title="Permalink to this headline"></a></h2> + <p>The Streams Rebalance Protocol offers several key advantages over the classic client-driven protocol:</p> + + <ul class="simple"> + <li><strong>Broker-Driven Coordination</strong>: Centralizes task assignment logic on brokers instead of the client. This provides consistent, authoritative task assignment decisions from a single coordination point and reduces the potential for split-brain scenarios.</li> + + <li><strong>Faster, More Stable Rebalances</strong>: Reduces rebalance duration and impact by removing the global synchronization point. This minimizes application downtime during membership changes or failures.</li> + + <li><strong>Better Observability</strong>: Provides dedicated metrics and admin interfaces that separate streams from consumer groups, leading to clearer troubleshooting with broker-side observability. See the <a class="reference external" href="/{{version}}/documentation.html#group_coordinator_monitoring">streams groups metrics</a> documentation for details.</li> + </ul> + </div> + + <div class="section" id="enabling"> + <h2><a class="toc-backref" href="#id5">Enabling the Protocol</a><a class="headerlink" href="#enabling" title="Permalink to this headline"></a></h2> + <p>The Streams Rebalance Protocol is enabled by default on new clusters starting with Apache Kafka 4.2. Both brokers and clients must be running Apache Kafka 4.2 or later to use this protocol.</p> + + <h3>Broker Configuration</h3> + <p>The protocol is enabled by default on new Apache Kafka 4.2 clusters. To enable the feature on existing clusters or to explicitly control it:</p> + + <p>Enable the feature:</p> + <pre><code>kafka-features.sh --bootstrap-server localhost:9092 upgrade --feature streams.version=1</code></pre> + + <p>Disable the feature:</p> + <pre><code>kafka-features.sh --bootstrap-server localhost:9092 downgrade --feature streams.version=0</code></pre> + + <h3>Client Configuration</h3> + <p>In your Kafka Streams application configuration, set:</p> + <pre><code>group.protocol=streams</code></pre> + </div> + + <div class="section" id="configuration"> + <h2><a class="toc-backref" href="#id6">Configuration</a><a class="headerlink" href="#configuration" title="Permalink to this headline"></a></h2> + + <div class="section" id="broker-configuration"> + <h3><a class="toc-backref" href="#id7">Broker Configuration</a><a class="headerlink" href="#broker-configuration" title="Permalink to this headline"></a></h3> + <p>The following broker configurations control the behavior of streams groups. For complete details, see the <a class="reference external" href="/{{version}}/documentation.html#brokerconfigs">broker configuration</a> documentation.</p> + + <ul class="simple"> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.coordinator.rebalance.protocols"><code>group.coordinator.rebalance.protocols</code></a>: The list of enabled rebalance protocols. <code>"streams"</code> is included in the list of protocols to enable streams groups.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.session.timeout.ms"><code>group.streams.session.timeout.ms</code></a>: The timeout to detect client failures when using the streams group protocol.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.min.session.timeout.ms"><code>group.streams.min.session.timeout.ms</code></a>: The minimum session timeout.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.max.session.timeout.ms"><code>group.streams.max.session.timeout.ms</code></a>: The maximum session timeout.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.heartbeat.interval.ms"><code>group.streams.heartbeat.interval.ms</code></a>: The heartbeat interval given to the members.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.min.heartbeat.interval.ms"><code>group.streams.min.heartbeat.interval.ms</code></a>: The minimum heartbeat interval.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.max.heartbeat.interval.ms"><code>group.streams.max.heartbeat.interval.ms</code></a>: The maximum heartbeat interval.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.max.size"><code>group.streams.max.size</code></a>: The maximum number of streams clients that a single streams group can accommodate.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.num.standby.replicas"><code>group.streams.num.standby.replicas</code></a>: The number of standby replicas for each task.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.max.standby.replicas"><code>group.streams.max.standby.replicas</code></a>: Maximum for dynamic configurations of the standby replica configuration.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.initial.rebalance.delay.ms"><code>group.streams.initial.rebalance.delay.ms</code></a>: The first rebalance of a group is delayed by this amount to allow more members to join the group.</li> + </ul> + </div> + + <div class="section" id="group-configuration"> + <h3><a class="toc-backref" href="#id8">Group Configuration</a><a class="headerlink" href="#group-configuration" title="Permalink to this headline"></a></h3> + <p>Configurations for the resource type <code>GROUP</code> are available in <code>DescribeConfigs</code> and <code>IncrementalAlterConfigs</code> to override the default broker configurations dynamically for specific groups. These can be set using the <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/admin/Admin.html">Admin</a> Java interface or the <code>kafka-configs.sh</code> utility.</p> + + <p>For complete details, see the <a class="reference external" href="/{{version}}/documentation.html#groupconfigs">group configuration</a> documentation.</p> + + <p>The following group-level configurations are available for streams groups:</p> + + <ul class="simple"> + <li><a class="reference external" href="/{{version}}/documentation/#groupconfigs_streams.session.timeout.ms"><code>streams.session.timeout.ms</code></a>: The timeout to detect client failures when using the streams group protocol.</li> + <li><a class="reference external" href="/{{version}}/documentation/#groupconfigs_streams.heartbeat.interval.ms"><code>streams.heartbeat.interval.ms</code></a>: The heartbeat interval given to the members.</li> + <li><a class="reference external" href="/{{version}}/documentation/#groupconfigs_streams.num.standby.replicas"><code>streams.num.standby.replicas</code></a>: The number of standby replicas for each task.</li> + <li><a class="reference external" href="/{{version}}/documentation/#groupconfigs_streams.initial.rebalance.delay.ms"><code>streams.initial.rebalance.delay.ms</code></a>: The first rebalance of a group is delayed by this amount to allow more members to join the group.</li> + </ul> + + <h4>Example: Setting Group-Level Configuration</h4> + <pre><code>kafka-configs.sh --bootstrap-server localhost:9092 \ + --alter --entity-type groups --entity-name wordcount \ + --add-config streams.num.standby.replicas=1</code></pre> + + <div class="admonition note"> + <strong>Note:</strong> In the streams rebalance protocol, <code>session.timeout.ms</code>, <code>heartbeat.interval.ms</code> and <code>num.standby.replicas</code> are group-level configurations, which are ignored when they are set on the client side. Use the <code>kafka-configs.sh</code> tool to set these configurations as shown above. + </div> + </div> + + <div class="section" id="streams-configuration"> + <h3><a class="toc-backref" href="#id9">Streams Configuration</a><a class="headerlink" href="#streams-configuration" title="Permalink to this headline"></a></h3> + <p>For complete details on all Kafka Streams configurations, see the <a class="reference external" href="/{{version}}/documentation.html#streamsconfigs">streams configuration</a> documentation.</p> + + <p>The following client configuration enables the streams rebalance protocol:</p> + + <ul class="simple"> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_group.protocol"><code>group.protocol</code></a>: A flag which indicates if the streams rebalance protocol should be used. Set to <code>streams</code> to enable (default is <code>classic</code>).</li> + </ul> + + <h4>Ignored Configurations</h4> + <p>The following configurations are ignored when the streams rebalance protocol is enabled:</p> + <ul class="simple"> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_acceptable.recovery.lag"><code>acceptable.recovery.lag</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_max.warmup.replicas"><code>max.warmup.replicas</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_num.standby.replicas"><code>num.standby.replicas</code></a> (use group-level configuration instead)</li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_probing.rebalance.interval.ms"><code>probing.rebalance.interval.ms</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_rack.aware.assignment.tags"><code>rack.aware.assignment.tags</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_rack.aware.assignment.strategy"><code>rack.aware.assignment.strategy</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_rack.aware.assignment.traffic_cost"><code>rack.aware.assignment.traffic_cost</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_rack.aware.assignment.non_overlap_cost"><code>rack.aware.assignment.non_overlap_cost</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_task.assignor.class"><code>task.assignor.class</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_session.timeout.ms"><code>session.timeout.ms</code></a> (use group-level configuration instead)</li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_heartbeat.interval.ms"><code>heartbeat.interval.ms</code></a> (use group-level configuration instead)</li> + </ul> + </div> + </div> + + <div class="section" id="administration"> + <h2><a class="toc-backref" href="#id10">Administration</a><a class="headerlink" href="#administration" title="Permalink to this headline"></a></h2> + + <h3>Admin API</h3> + <p>Use the "streams groups" methods of the <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/admin/Admin.html">org.apache.kafka.clients.admin.Admin</a> interface to manage streams groups programmatically. These APIs are mostly backed by the same implementations as the consumer group API.</p> + + <p>The main differences from consumer group APIs are:</p> + <ul class="simple"> + <li>The <code>describeStreamsGroups</code> uses the DescribeStreamsGroup RPC and contains different information than consumer groups.</li> + <li>A streams group has an extra state - <code>NOT_READY</code> - and no legacy states from the classic protocol.</li> + <li><code>removeMembersFromConsumerGroup</code> will not have a corresponding API in this version, as it uses the LeaveGroup RPC for classic consumer groups, which is not available for KIP-848-style groups.</li> + </ul> + + <h3>kafka-streams-groups.sh</h3> + <p>A new tool called <a class="reference internal" href="kafka-streams-group-sh.html">kafka-streams-groups.sh</a> is added for working with streams groups. It replaces <code>kafka-streams-application-reset</code> for streams groups and can be used to list, describe, and delete streams groups. See the <a class="reference internal" href="kafka-streams-group-sh.html">kafka-streams-groups.sh documentation</a> for detailed usage information.</p> + </div> + + <div class="section" id="architecture"> + <h2><a class="toc-backref" href="#id11">Architecture and How It Works</a><a class="headerlink" href="#architecture" title="Permalink to this headline"></a></h2> + + <div class="section" id="streams-groups"> + <h3><a class="toc-backref" href="#id12">Streams Groups</a><a class="headerlink" href="#streams-groups" title="Permalink to this headline"></a></h3> + <p>The protocol introduces the concept of a <strong>streams group</strong> in parallel to a consumer group. Streams clients use a dedicated heartbeat RPC, <code>StreamsGroupHeartbeat</code>, to join a group, leave a group, and update the group coordinator about its currently owned tasks and its client-specific metadata.</p> + + <p>The group coordinator manages a streams group similarly to a consumer group, continuously updating the group member metadata via heartbeat responses and running assignment logic when changes are detected. A new group type called <code>streams</code> is introduced to the group coordinator, with new record key and value types for group metadata, topology metadata, and group member metadata. These records are persisted in the <code>__consumer_offsets</code> topic.</p> + + <p>A group can either be a streams group, a share group, or a consumer group, defined by the first heartbeat request using the corresponding GroupId.</p> + </div> + + <div class="section" id="topology-configuration"> + <h3><a class="toc-backref" href="#id13">Topology Configuration and Validation</a><a class="headerlink" href="#topology-configuration" title="Permalink to this headline"></a></h3> + <p>To assign tasks among streams clients, the group coordinator uses topology metadata that is initialized when a member joins the group and persisted in the consumer offsets topic.</p> + + <p>Whenever a member joins the streams group, the first heartbeat request contains metadata of the topology. The metadata describes the topology as a set of subtopologies, each identified by a unique string identifier and containing metadata relevant for creation of internal topics and assignment.</p> Review Comment: The term should be "sub-topologies" (with hyphen) to be consistent with the rest of the Kafka Streams documentation. While "subtopology" (without hyphen) is used in code contexts like method names, in documentation text describing the topology structure, the hyphenated form "sub-topology/sub-topologies" is consistently used (see docs/streams/tutorial.html:504-505, docs/streams/developer-guide/dsl-topology-naming.html). ########## docs/streams/developer-guide/streams-rebalance-protocol.html: ########## @@ -0,0 +1,351 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +<script><!--#include virtual="../../js/templateData.js" --></script> + +<script id="content-template" type="text/x-handlebars-template"> + <!-- h1>Developer Guide for Kafka Streams</h1 --> + <div class="sub-nav-sticky"> + <div class="sticky-top"> + <!-- div style="height:35px"> + <a href="/{{version}}/documentation/streams/">Introduction</a> + <a class="active-menu-item" href="/{{version}}/documentation/streams/developer-guide">Developer Guide</a> + <a href="/{{version}}/documentation/streams/core-concepts">Concepts</a> + <a href="/{{version}}/documentation/streams/quickstart">Run Demo App</a> + <a href="/{{version}}/documentation/streams/tutorial">Tutorial: Write App</a> + </div --> + </div> + </div> + + <div class="section" id="streams-rebalance-protocol"> + <h1>Streams Rebalance Protocol<a class="headerlink" href="#streams-rebalance-protocol" title="Permalink to this headline"></a></h1> + <p>The Streams Rebalance Protocol is a broker-driven rebalancing system designed specifically for Kafka Streams applications. Following the pattern of KIP-848, which moved rebalance coordination of plain consumers from clients to brokers, KIP-1071 extends this model to Kafka Streams workloads.</p> + + <div class="contents local topic" id="table-of-contents"> + <p class="topic-title first"><b>Table of Contents</b></p> + <ul class="simple"> + <li><a class="reference internal" href="#overview" id="id1">Overview</a></li> + <li><a class="reference internal" href="#whats-supported" id="id2">What's Supported in This Version</a></li> + <li><a class="reference internal" href="#whats-not-supported" id="id3">What's Not Supported in This Version</a></li> + <li><a class="reference internal" href="#why-use" id="id4">Why Use the Streams Rebalance Protocol?</a></li> + <li><a class="reference internal" href="#enabling" id="id5">Enabling the Protocol</a></li> + <li><a class="reference internal" href="#configuration" id="id6">Configuration</a><ul> + <li><a class="reference internal" href="#broker-configuration" id="id7">Broker Configuration</a></li> + <li><a class="reference internal" href="#group-configuration" id="id8">Group Configuration</a></li> + <li><a class="reference internal" href="#streams-configuration" id="id9">Streams Configuration</a></li> + </ul> + </li> + <li><a class="reference internal" href="#administration" id="id10">Administration</a></li> + <li><a class="reference internal" href="#architecture" id="id11">Architecture and How It Works</a><ul> + <li><a class="reference internal" href="#streams-groups" id="id12">Streams Groups</a></li> + <li><a class="reference internal" href="#topology-configuration" id="id13">Topology Configuration and Validation</a></li> + <li><a class="reference internal" href="#assignment-configuration" id="id14">Centralized Assignment Configuration</a></li> + </ul> + </li> + <li><a class="reference internal" href="#monitoring" id="id15">Monitoring and Metrics</a></li> + <li><a class="reference internal" href="#migration" id="id16">Migration from Classic Protocol</a></li> + </ul> + </div> + + <div class="section" id="overview"> + <h2><a class="toc-backref" href="#id1">Overview</a><a class="headerlink" href="#overview" title="Permalink to this headline"></a></h2> + <p>Instead of clients computing new assignments on the client during rebalance events involving all members of the group, assignments are computed continuously on the broker. Instead of using a consumer group, the streams application registers as a <strong>streams group</strong> with the broker, which manages and exposes all metadata required for coordination of the streams application instances.</p> + + <p>This approach brings Kafka Streams coordination in line with the modern broker-driven rebalance model introduced for consumers in KIP-848, providing a dedicated group type with streams-specific semantics and metadata management.</p> + </div> + + <div class="section" id="whats-supported"> + <h2><a class="toc-backref" href="#id2">What's Supported in This Version</a><a class="headerlink" href="#whats-supported" title="Permalink to this headline"></a></h2> + <p>The following features are available in the current release:</p> + <ul class="simple"> + <li><strong>Core Streams Group Rebalance Protocol</strong>: The <code>group.protocol=streams</code> configuration enables the dedicated streams rebalance protocol. This separates streams groups from consumer groups and provides a streams-specific group membership lifecycle and metadata management on the broker.</li> + + <li><strong>Sticky Task Assignor</strong>: A basic task assignment strategy that minimizes task movement during rebalances is included.</li> + + <li><strong>Interactive Query Support</strong>: IQ operations are compatible with the new streams protocol.</li> + + <li><strong>New Admin RPC</strong>: The StreamsGroupDescribe RPC provides streams-specific metadata separate from consumer group information, with corresponding access via the <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/admin/Admin.html">Admin client</a>.</li> + + <li><strong>CLI Integration</strong>: You can list, describe, and delete streams groups via the <a class="reference internal" href="kafka-streams-group-sh.html">kafka-streams-groups.sh</a> script.</li> + + <li><strong>Offline Migration</strong>: After shutting down all members and waiting for their <code>session.timeout.ms</code> to expire, a classic group can be converted to a streams group and a streams group can be converted to a classic group. The only thing that will be preserved broker-side are the committed offsets of the application.</li> + </ul> + </div> + + <div class="section" id="whats-not-supported"> + <h2><a class="toc-backref" href="#id3">What's Not Supported in This Version</a><a class="headerlink" href="#whats-not-supported" title="Permalink to this headline"></a></h2> + <p>The following features are not yet available and should be avoided when using the new protocol:</p> + <ul class="simple"> + <li><strong>Static Membership</strong>: Setting a client <code>instance.id</code> will be rejected.</li> + + <li><strong>Topology Updates</strong>: If a topology is changed significantly (e.g., by adding new source topics or changing the number of sub-topologies), a new streams group must be created.</li> + + <li><strong>High Availability Assignor</strong>: Only the sticky assignor is supported.</li> + + <li><strong>Regular Expressions</strong>: Pattern-based topic subscription is not supported.</li> + + <li><strong>Online Migration</strong>: Group migration while the application is running is not available between the classic and new streams protocol.</li> + </ul> + </div> + + <div class="section" id="why-use"> + <h2><a class="toc-backref" href="#id4">Why Use the Streams Rebalance Protocol?</a><a class="headerlink" href="#why-use" title="Permalink to this headline"></a></h2> + <p>The Streams Rebalance Protocol offers several key advantages over the classic client-driven protocol:</p> + + <ul class="simple"> + <li><strong>Broker-Driven Coordination</strong>: Centralizes task assignment logic on brokers instead of the client. This provides consistent, authoritative task assignment decisions from a single coordination point and reduces the potential for split-brain scenarios.</li> + + <li><strong>Faster, More Stable Rebalances</strong>: Reduces rebalance duration and impact by removing the global synchronization point. This minimizes application downtime during membership changes or failures.</li> + + <li><strong>Better Observability</strong>: Provides dedicated metrics and admin interfaces that separate streams from consumer groups, leading to clearer troubleshooting with broker-side observability. See the <a class="reference external" href="/{{version}}/documentation.html#group_coordinator_monitoring">streams groups metrics</a> documentation for details.</li> + </ul> + </div> + + <div class="section" id="enabling"> + <h2><a class="toc-backref" href="#id5">Enabling the Protocol</a><a class="headerlink" href="#enabling" title="Permalink to this headline"></a></h2> + <p>The Streams Rebalance Protocol is enabled by default on new clusters starting with Apache Kafka 4.2. Both brokers and clients must be running Apache Kafka 4.2 or later to use this protocol.</p> + + <h3>Broker Configuration</h3> + <p>The protocol is enabled by default on new Apache Kafka 4.2 clusters. To enable the feature on existing clusters or to explicitly control it:</p> + + <p>Enable the feature:</p> + <pre><code>kafka-features.sh --bootstrap-server localhost:9092 upgrade --feature streams.version=1</code></pre> + + <p>Disable the feature:</p> + <pre><code>kafka-features.sh --bootstrap-server localhost:9092 downgrade --feature streams.version=0</code></pre> + + <h3>Client Configuration</h3> + <p>In your Kafka Streams application configuration, set:</p> + <pre><code>group.protocol=streams</code></pre> + </div> + + <div class="section" id="configuration"> + <h2><a class="toc-backref" href="#id6">Configuration</a><a class="headerlink" href="#configuration" title="Permalink to this headline"></a></h2> + + <div class="section" id="broker-configuration"> + <h3><a class="toc-backref" href="#id7">Broker Configuration</a><a class="headerlink" href="#broker-configuration" title="Permalink to this headline"></a></h3> + <p>The following broker configurations control the behavior of streams groups. For complete details, see the <a class="reference external" href="/{{version}}/documentation.html#brokerconfigs">broker configuration</a> documentation.</p> + + <ul class="simple"> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.coordinator.rebalance.protocols"><code>group.coordinator.rebalance.protocols</code></a>: The list of enabled rebalance protocols. <code>"streams"</code> is included in the list of protocols to enable streams groups.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.session.timeout.ms"><code>group.streams.session.timeout.ms</code></a>: The timeout to detect client failures when using the streams group protocol.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.min.session.timeout.ms"><code>group.streams.min.session.timeout.ms</code></a>: The minimum session timeout.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.max.session.timeout.ms"><code>group.streams.max.session.timeout.ms</code></a>: The maximum session timeout.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.heartbeat.interval.ms"><code>group.streams.heartbeat.interval.ms</code></a>: The heartbeat interval given to the members.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.min.heartbeat.interval.ms"><code>group.streams.min.heartbeat.interval.ms</code></a>: The minimum heartbeat interval.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.max.heartbeat.interval.ms"><code>group.streams.max.heartbeat.interval.ms</code></a>: The maximum heartbeat interval.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.max.size"><code>group.streams.max.size</code></a>: The maximum number of streams clients that a single streams group can accommodate.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.num.standby.replicas"><code>group.streams.num.standby.replicas</code></a>: The number of standby replicas for each task.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.max.standby.replicas"><code>group.streams.max.standby.replicas</code></a>: Maximum for dynamic configurations of the standby replica configuration.</li> + <li><a class="reference external" href="/{{version}}/documentation/#brokerconfigs_group.streams.initial.rebalance.delay.ms"><code>group.streams.initial.rebalance.delay.ms</code></a>: The first rebalance of a group is delayed by this amount to allow more members to join the group.</li> + </ul> + </div> + + <div class="section" id="group-configuration"> + <h3><a class="toc-backref" href="#id8">Group Configuration</a><a class="headerlink" href="#group-configuration" title="Permalink to this headline"></a></h3> + <p>Configurations for the resource type <code>GROUP</code> are available in <code>DescribeConfigs</code> and <code>IncrementalAlterConfigs</code> to override the default broker configurations dynamically for specific groups. These can be set using the <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/admin/Admin.html">Admin</a> Java interface or the <code>kafka-configs.sh</code> utility.</p> + + <p>For complete details, see the <a class="reference external" href="/{{version}}/documentation.html#groupconfigs">group configuration</a> documentation.</p> + + <p>The following group-level configurations are available for streams groups:</p> + + <ul class="simple"> + <li><a class="reference external" href="/{{version}}/documentation/#groupconfigs_streams.session.timeout.ms"><code>streams.session.timeout.ms</code></a>: The timeout to detect client failures when using the streams group protocol.</li> + <li><a class="reference external" href="/{{version}}/documentation/#groupconfigs_streams.heartbeat.interval.ms"><code>streams.heartbeat.interval.ms</code></a>: The heartbeat interval given to the members.</li> + <li><a class="reference external" href="/{{version}}/documentation/#groupconfigs_streams.num.standby.replicas"><code>streams.num.standby.replicas</code></a>: The number of standby replicas for each task.</li> + <li><a class="reference external" href="/{{version}}/documentation/#groupconfigs_streams.initial.rebalance.delay.ms"><code>streams.initial.rebalance.delay.ms</code></a>: The first rebalance of a group is delayed by this amount to allow more members to join the group.</li> + </ul> + + <h4>Example: Setting Group-Level Configuration</h4> + <pre><code>kafka-configs.sh --bootstrap-server localhost:9092 \ + --alter --entity-type groups --entity-name wordcount \ + --add-config streams.num.standby.replicas=1</code></pre> + + <div class="admonition note"> + <strong>Note:</strong> In the streams rebalance protocol, <code>session.timeout.ms</code>, <code>heartbeat.interval.ms</code> and <code>num.standby.replicas</code> are group-level configurations, which are ignored when they are set on the client side. Use the <code>kafka-configs.sh</code> tool to set these configurations as shown above. + </div> + </div> + + <div class="section" id="streams-configuration"> + <h3><a class="toc-backref" href="#id9">Streams Configuration</a><a class="headerlink" href="#streams-configuration" title="Permalink to this headline"></a></h3> + <p>For complete details on all Kafka Streams configurations, see the <a class="reference external" href="/{{version}}/documentation.html#streamsconfigs">streams configuration</a> documentation.</p> + + <p>The following client configuration enables the streams rebalance protocol:</p> + + <ul class="simple"> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_group.protocol"><code>group.protocol</code></a>: A flag which indicates if the streams rebalance protocol should be used. Set to <code>streams</code> to enable (default is <code>classic</code>).</li> + </ul> + + <h4>Ignored Configurations</h4> + <p>The following configurations are ignored when the streams rebalance protocol is enabled:</p> + <ul class="simple"> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_acceptable.recovery.lag"><code>acceptable.recovery.lag</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_max.warmup.replicas"><code>max.warmup.replicas</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_num.standby.replicas"><code>num.standby.replicas</code></a> (use group-level configuration instead)</li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_probing.rebalance.interval.ms"><code>probing.rebalance.interval.ms</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_rack.aware.assignment.tags"><code>rack.aware.assignment.tags</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_rack.aware.assignment.strategy"><code>rack.aware.assignment.strategy</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_rack.aware.assignment.traffic_cost"><code>rack.aware.assignment.traffic_cost</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_rack.aware.assignment.non_overlap_cost"><code>rack.aware.assignment.non_overlap_cost</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_task.assignor.class"><code>task.assignor.class</code></a></li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_session.timeout.ms"><code>session.timeout.ms</code></a> (use group-level configuration instead)</li> + <li><a class="reference external" href="/{{version}}/documentation/#streamsconfigs_heartbeat.interval.ms"><code>heartbeat.interval.ms</code></a> (use group-level configuration instead)</li> + </ul> + </div> + </div> + + <div class="section" id="administration"> + <h2><a class="toc-backref" href="#id10">Administration</a><a class="headerlink" href="#administration" title="Permalink to this headline"></a></h2> + + <h3>Admin API</h3> + <p>Use the "streams groups" methods of the <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/admin/Admin.html">org.apache.kafka.clients.admin.Admin</a> interface to manage streams groups programmatically. These APIs are mostly backed by the same implementations as the consumer group API.</p> + + <p>The main differences from consumer group APIs are:</p> + <ul class="simple"> + <li>The <code>describeStreamsGroups</code> uses the DescribeStreamsGroup RPC and contains different information than consumer groups.</li> + <li>A streams group has an extra state - <code>NOT_READY</code> - and no legacy states from the classic protocol.</li> + <li><code>removeMembersFromConsumerGroup</code> will not have a corresponding API in this version, as it uses the LeaveGroup RPC for classic consumer groups, which is not available for KIP-848-style groups.</li> + </ul> + + <h3>kafka-streams-groups.sh</h3> + <p>A new tool called <a class="reference internal" href="kafka-streams-group-sh.html">kafka-streams-groups.sh</a> is added for working with streams groups. It replaces <code>kafka-streams-application-reset</code> for streams groups and can be used to list, describe, and delete streams groups. See the <a class="reference internal" href="kafka-streams-group-sh.html">kafka-streams-groups.sh documentation</a> for detailed usage information.</p> + </div> + + <div class="section" id="architecture"> + <h2><a class="toc-backref" href="#id11">Architecture and How It Works</a><a class="headerlink" href="#architecture" title="Permalink to this headline"></a></h2> + + <div class="section" id="streams-groups"> + <h3><a class="toc-backref" href="#id12">Streams Groups</a><a class="headerlink" href="#streams-groups" title="Permalink to this headline"></a></h3> + <p>The protocol introduces the concept of a <strong>streams group</strong> in parallel to a consumer group. Streams clients use a dedicated heartbeat RPC, <code>StreamsGroupHeartbeat</code>, to join a group, leave a group, and update the group coordinator about its currently owned tasks and its client-specific metadata.</p> + + <p>The group coordinator manages a streams group similarly to a consumer group, continuously updating the group member metadata via heartbeat responses and running assignment logic when changes are detected. A new group type called <code>streams</code> is introduced to the group coordinator, with new record key and value types for group metadata, topology metadata, and group member metadata. These records are persisted in the <code>__consumer_offsets</code> topic.</p> + + <p>A group can either be a streams group, a share group, or a consumer group, defined by the first heartbeat request using the corresponding GroupId.</p> + </div> + + <div class="section" id="topology-configuration"> + <h3><a class="toc-backref" href="#id13">Topology Configuration and Validation</a><a class="headerlink" href="#topology-configuration" title="Permalink to this headline"></a></h3> + <p>To assign tasks among streams clients, the group coordinator uses topology metadata that is initialized when a member joins the group and persisted in the consumer offsets topic.</p> + + <p>Whenever a member joins the streams group, the first heartbeat request contains metadata of the topology. The metadata describes the topology as a set of subtopologies, each identified by a unique string identifier and containing metadata relevant for creation of internal topics and assignment.</p> + + <h4>Topology Validation and NOT_READY State</h4> + <p>During the handling of the streams group heartbeat, the group coordinator may detect that source/sink or internal topics required by the topology do not exist or differ in their configuration from what is required for the topology to execute successfully. This triggers a "topology configuration" process, in which the group coordinator performs the following steps:</p> + + <ul class="simple"> + <li>Check that all source topics exist and resolve source topic regular expressions (checking that each resolves to at least one topic).</li> + <li>Check that "copartition groups" are satisfied - that is, all source topics that are supposed to be copartitioned are indeed copartitioned.</li> Review Comment: The term should be "co-partitioned" (with hyphen) to be consistent with the rest of the Kafka Streams documentation. The codebase consistently uses "co-partitioned" throughout other documentation files. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
