DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
Status
Current state: accepted
Discussion thread: here
Voting thread: here
JIRA: here
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
Kafka Streams supports custom task assignors in the classic client-side assignment model via task.assignor.class. However, this customization does not carry over to the Streams Rebalance Protocol (KIP-1071), where task assignment is computed on the broker rather than in the client. As a result, users of group.protocol=streams cannot provide a custom assignor implementation, even though a similar capability exists both in classic Streams and in KIP-848 consumer groups (via group.consumer.assignors).
This KIP closes the feature gap by adding first-class broker-side custom task assignor support for streams groups. Specifically, it:
- Exposes the necessary public Java interfaces so that users can implement custom task assignors.
- Supports configuring assignors through a static broker configuration
group.streams.assignorsthat accepts both built-in assignor short names and fully qualified class names of custom implementations. - Provides a per-group configuration to select an assignor by short name, with the broker-level default determined by the first entry in the
group.streams.assignorslist. - Preserves existing KIP-1071 behavior when no custom assignor is configured.
Public Interfaces
New Public Java Interfaces
The following types are moved from the internal group-coordinator module package org.apache.kafka.coordinator.group.streams.assignor to the public group-coordinator-api module, under the new package org.apache.kafka.coordinator.group.api.streams.assignor. A dedicated sub-package is used to avoid name collisions with the existing consumer group assignor types in org.apache.kafka.coordinator.group.api.assignor.
These types exist today as internal classes, and this KIP makes them public API with the signatures below. All newly public types are annotated @InterfaceStability.Evolving. Following the KIP-848 convention, the input and output container types are defined as interfaces rather than records, so the API can evolve in future releases. AssignmentConfigs is the one exception to the above: it is a new interface introduced by this KIP rather than a relocated internal type.
Entry point
package org.apache.kafka.coordinator.group.api.streams.assignor;
/**
* Server-side task assignor used by streams groups.
* <p>Implementations must be thread-safe: a single instance is shared across all
* streams groups on a broker.
*
* An implementation may also implement {@link org.apache.kafka.common.Configurable}.
* If it does, the broker invokes {@code configure(Map)} once with the broker
* configuration when the assignor is loaded,
* following the standard pattern for broker-side plugins.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public interface TaskAssignor {
/**
* Unique short name for this assignor. Used in configurations to select this assignor.
*/
String name();
/**
* Assigns tasks to group members based on the given assignment specification
* and topology metadata.
*
* @param groupSpec The assignment spec which includes member metadata.
* @param topologyDescriber The topology and task metadata describer.
* @return The new assignment for the group.
*
* @throws TaskAssignorException on assignment failure.
*/
GroupAssignment assign(
GroupSpec groupSpec,
TopologyDescriber topologyDescriber
) throws TaskAssignorException;
}
Input types (parameters to assign())
AssignmentConfigs carries the assignment-relevant group configurations. The group coordinator resolves every value from the broker configuration and the per-group override before invoking the assignor, so an assignor always sees a fully resolved value and never has to handle a missing configuration. The two config currently we have now are numStandbyReplicas() comes from group.streams.num.standby.replicas, overridden per group by streams.num.standby.replicas (0 by default), and rackAwareAssignmentTags() comes from group.streams.rack.aware.assignment.tags, overridden per group by streams.rack.aware.assignment.tags (empty by default).
/**
* The group metadata specification required to compute the target assignment.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public interface GroupSpec {
/**
* @return The member IDs in the group.
*/
Collection<String> memberIds();
/**
* @return The static metadata for the given member.
*/
MemberAssignmentMetadata memberMetadata(String memberId);
/**
* @return The current task assignment state for the given member
*/
MemberAssignmentState memberAssignmentState(String memberId);
/**
* @return The assignment configurations passed to the assignor.
*/
AssignmentConfigs configs();
}
/**
* Static, per-member metadata that does not change as a result of assignment.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public interface MemberAssignmentMetadata {
Optional<String> instanceId();
Optional<String> rackId();
String processId();
Map<String, String> clientTags();
}
/**
* A member's current task assignment state, provided as input to the assignor.
* Warm-up tasks are included here because a member may currently be warming up.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public interface MemberAssignmentState {
/** Current target active tasks, keyed by subtopology ID. */
Map<String, Set<Integer>> activeTasks();
/** Current target standby tasks, keyed by subtopology ID. */
Map<String, Set<Integer>> standbyTasks();
/** Current warmup tasks, keyed by subtopology ID. */
Map<String, Set<Integer>> warmupTasks();
/** The last received cumulative task offsets of assigned or dormant tasks. */
Map<String, Map<Integer, Long>> taskOffsets();
/** The last received task end offsets. */
Map<String, Map<Integer, Long>> taskEndOffsets();
}
/**
* The assignment configurations that the group coordinator passes to the task assignor.
*
* <p>This interface is not intended to be implemented by task assignors: new configurations may be added to it.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public interface AssignmentConfigs {
/**
* @return The number of standby replicas for each task.
*/
int numStandbyReplicas();
/**
* @return The client tags used to distribute standby tasks across racks. The list is unmodifiable.
*/
List<String> rackAwareAssignmentTags();
}
/**
* Used by the assignor to get topic and task metadata of the group's topology.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public interface TopologyDescriber {
/**
* @return The list of subtopology IDs.
*/
List<String> subtopologies();
/**
* The maximal number of input partitions among all source topics for the given subtopology.
*
* @param subtopologyId String identifying the subtopology.
* @throws NoSuchElementException if the subtopology ID does not exist.
* @throws IllegalStateException if the subtopology contains no source topics.
*/
int maxNumInputPartitions(String subtopologyId);
/**
* Whether the given subtopology is associated with a changelog topic.
*
* @param subtopologyId String identifying the subtopology.
* @throws NoSuchElementException if the subtopology ID does not exist.
*/
boolean isStateful(String subtopologyId);
}
Output types (returned from assign())
/**
* The task assignment for a streams group. Constructed and returned by the assignor.
*/
@InterfaceAudience.Public @InterfaceStability.Evolving
public class GroupAssignment {
private final Map<String, MemberAssignment> members;
public GroupAssignment(Map<String, MemberAssignment> members) {
this.members = Objects.requireNonNull(members);
}
public Map<String, MemberAssignment> members() {
return members;
}
// equals / hashCode / toString
}
/**
* The task assignment computed by the assignor for a streams group member.
*
* Note: warm-up tasks are intentionally NOT part of the assignor output. The
* assignor computes only active and standby tasks; warm-up tasks are derived by
* the group coordinator's reconciler.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public class MemberAssignment {
public MemberAssignment(Map<String, Set<Integer>> activeTasks, Map<String, Set<Integer>> standbyTasks);
/** @return The active tasks assigned to this member keyed by subtopology Id. */
public Map<String, Set<Integer>> activeTasks();
/** @return The standby tasks assigned to this member keyed by subtopology Id. */
public Map<String, Set<Integer>> standbyTasks();
// equals(), hashCode(), toString() are also provided
}
Supporting types
/**
* Exception thrown by {@link TaskAssignor#assign(GroupSpec, TopologyDescriber)} on
* assignment failure.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public class TaskAssignorException extends ApiException {
public TaskAssignorException(String message);
public TaskAssignorException(String message, Throwable cause);
}
Built-in assignors
The sticky task assignor is the only assignor shipped with the broker, and it minimizes task movement during rebalances. It is registered under the short name sticky, while its fully qualified class name is org.apache.kafka.coordinator.group.streams.assignor.StickyTaskAssignor. Both forms are public API and may be used in group.streams.assignors, and the class name will not change without a KIP. This follows the existing convention for consumer groups and share groups which established in KIP-848. The class itself stays in the module where it lives today rather than moving to the public API module, which is consistent with the assignors shipped for consumer and share groups. Only the class name is public API, and the class inot intended to be subclassed or referenced from user code.
New broker configurations
group.streams.assignors
| Type | LIST of assignor short names |
| Default | [sticky] |
| Importance | Medium |
| Update mode | Static (read-only, requires broker restart) |
The server-side task assignors for streams groups, as a list of either short names for built-in assignors or fully qualified class names for custom assignors. Each entry is resolved to a TaskAssignor instance at broker startup; if the instance implements org.apache.kafka.common.Configurable, configure() is invoked with the broker configuration. Assignors are keyed by the value returned by TaskAssignor#name(). The first entry in the list determines the assignor used by groups that do not set the streams.assignor.name group configuration. The default value is sticky. Duplicate names are not allowed, and any class that fails to load or whose name() collides with another assignor causes the broker to fail to start.
New group configuration
streams.assignor.name
| Type | STRING (assignor short name) |
| Default | null (use the first entry of group.streams.assignor) |
| Update mode | Dynamic per group, via the IncrementalAlterConfigs admin RPC |
Selects the task assignor for a specific streams group, by short name. The name must be present in the broker's group.streams.assignors list; otherwise the IncrementalAlterConfigs request is rejected with INVALID_CONFIG.
StreamsGroupDescribeResponse Change
One new field is added to each DescribedGroup at version 1. StreamsGroupDescribeResponse is not bumped by this KIP. Version 1 is introduced by KIP-1331 and has not been released yet, so AssignorName is added to that existing version and validVersions remains 0-1.
{ "name": "AssignorName", "type": "string", "versions": "1+", "nullableVersions": "1+",
"default": "null", "ignorable": true,
"about": "The task assignor that the group coordinator will use for the next assignment
computation. This may differ from the assignor that computed the current
assignment, because changing the assignor configuration does not trigger a
rebalance. Null in case of a describe error." }
The field reports the assignor for the next assignment computation rather than the one that produced the current assignment, because changing streams.assignor.name does not trigger a rebalance. There is no change to StreamsGroupDescribeRequest: the assignor name is always returned and does not need to be requested explicitly.
Admin API
StreamsGroupDescription exposes the assignor name public Optional<String> assignorName() The value is empty when the broker does not report one, for example an older broker answering at StreamsGroupDescribe version 0. assignorName is also added to the constructor. The constructor released in 4.2/4.3 is deprecated, and a single new constructor carries both the KIP-1331 topology description parameters and assignorName
@Deprecated(since = "4.4") public StreamsGroupDescription(String groupId, int groupEpoch, int targetAssignmentEpoch, int topologyEpoch, Collection<StreamsGroupSubtopologyDescription> subtopologies, Collection<StreamsGroupMemberDescription> members, GroupState groupState, Node coordinator, Set<AclOperation> authorizedOperations) public StreamsGroupDescription(String groupId, int groupEpoch, int targetAssignmentEpoch, int topologyEpoch, Collection<StreamsGroupSubtopologyDescription> subtopologies, Collection<StreamsGroupMemberDescription> members, GroupState groupState, Node coordinator, Set<AclOperation> authorizedOperations, Optional<StreamsGroupTopologyDescription> topologyDescription, StreamsGroupTopologyDescriptionStatus topologyDescriptionStatus, Optional<String> assignorName)
Proposed Changes
Assignor loading and startup validation
When the broker starts, it builds the registry of available task assignors by iterating group.streams.assignors in order. For each entry:
- If it is a built-in short name, the corresponding built-in assignor is registered.
- Otherwise it is treated as a fully qualified class name and instantiated via its public no-argument constructor.
- If the class cannot be loaded, the broker fails to start.
- If the instance implements
Configurable,configure(config.originals())is invoked with the broker configuration
- Register the assignor by
name(). If aname()collides with a built-in name or with another entry's name, the broker fails to start.
The result is a single registry Map<String, TaskAssignor> of all available assignors. Everything downstream operates on short names only. Misconfiguration in any step is fail-fast: the broker never starts with a partially valid assignor set.
Group configuration validation
When an operator sets streams.assignor.name for a group via IncrementalAlterConfigs:
- The receiving broker checks whether the assignor name exists in its registry of available assignors (built-in + custom).
- If the name is invalid, the broker rejects the request with
INVALID_CONFIG; the configuration is never forwarded to the controller. - If the name is valid, the request is forwarded to the controller, which persists it in the metadata log.
Pre-validation at the forwarding broker prevents invalid names from being persisted. This relies on all brokers having the same group.streams.assignors configuration, which is expected for consistent cluster behavior.
Assignor selection at assignment time
When the group coordinator computes a new target assignment for a streams group, it resolves the assignor as follows:
- If the group has
streams.assignor.nameset, use that assignor. - Otherwise, use the first entry of
group.streams.assignors
Runtime fallback
If an assignor name was previously persisted in a group's configuration but is no longer available on the coordinator's broker (for example, the operator removed a custom assignor from the broker configuration, or the group coordinator moved to a broker with a different configuration during a rolling restart):
- The group coordinator falls back to the first entry of
group.streams.assignors. - A warning is logged so the operator is aware of the mismatch.
- The group continues to function normally.
Assignment failure handling
If the selected assignor throws TaskAssignorException, the existing KIP-1071 behavior is preserved: the group's previous target assignment is kept unchanged, and the triggering heartbeat fails with UNKNOWN_SERVER_ERROR (the exception is rethrown as UnknownServerException, including the failure message). Assignment is retried on a subsequent rebalance trigger. The same applies to custom assignors; no new error code is introduced.
Compatibility, Deprecation, and Migration Plan
- Existing users are unaffected. With the default configuration
group.streams.assignors=[sticky]and no per-group override, the behavior of KIP-1071 streams groups is unchanged. - The relocated types are not a breaking change. They are internal today; moving them to the public group-coordinator-api module as @InterfaceStability.Evolving interfaces is safe, since the original package was never public API. The internal classes are removed. GroupSpec#configs() returns AssignmentConfigs instead of Map<String, String>. This is not a compatibility concern either, since GroupSpec is internal today and has never been part of a public API. Because task assignors are not expected to implement AssignmentConfigs, further assignment configurations can be exposed by adding methods to that interface in a later release without breaking existing custom assignors.The fully qualified class name of the built-in sticky assignor does become public API, as described under Public Interfaces.
- One deprecation in the Admin API. The 9-argument
StreamsGroupDescriptionconstructor released in 4.2/4.3 is deprecated for removal. The constructor added by KIP-1331 is extended in place with the assignorName argument, andStreamsGroupDescribeResponseversion 1 gains the AssignorName field in place; both in-place changes are safe because neither has been included in an Apache Kafka release. - The classic client-side
task.assignor.classis untouched. It continues to apply to the classic protocol only. Users migrating from the classic protocol togroup.protocol=streamswho rely on a custom client-side assignor will need to re-implement it against the new broker-side interface and have the cluster operator register it. - There are no new record formats and no new error codes. The per-group setting reuses the existing group configuration mechanism
IncrementalAlterConfigs. The only protocol change is a new optional AssignorName field inStreamsGroupDescribeResponse, which makes the effective assignor of a group observable.
Test Plan
- Unit tests for assignor loading and validation: class loading failures, name collisions, unknown names in
group.streams.assignors, duplicate handling, and group configuration validation inIncrementalAlterConfigs. - Integration tests that register a custom assignor on the broker, select it for a group via
streams.assignor.name, and verify the produced assignment is the custom assignor's output; plus the fallback path where a persisted assignor name is no longer available and the coordinator falls back to the default with a warning. - System tests running a Streams application under
group.protocol=streamswith a custom assignor configured, verifying correct task distribution, behavior across broker restarts (static config reload), and that existing system tests pass unchanged with the default configuration
Future work
During the discussion it was suggested to move validation of all dynamic group configs to the broker, so they are only validated in one place. This KIP only validates streams.assignor.name there, following the existing pattern. The larger cleanup is left as follow-up work.
Rejected Alternatives
Client-side assignor selection (KIP-848 style ServerAssignor voting)
KIP-848 consumer groups let each member suggest a server-side assignor via the ServerAssignor heartbeat field, with the coordinator picking the most-voted one. We rejected this for streams groups: KIP-1071 deliberately treats assignment as an operator concern, and task assignment quality depends on cluster-wide operational knowledge (standby placement, rack awareness) rather than per-client preference. Adding a heartbeat field would also require a wire protocol change for little benefit.
Splitting loading and selection into two broker configs
An earlier version of this proposal split assignor loading (group.streams.custom.assignor.classes) from selection (group.streams.assignors.names). This was rejected in favor of a single group.streams.assignors list that mixes built-in short names and fully qualified class names, consistent with group.consumer.assignors. Once assignors are loaded, their short names come from TaskAssignor#name(), which makes a separate names list redundant. A single list is simpler for operators and consistent with the existing consumer-group configuration.
Client-side custom assignment under the streams protocol
Keeping task.assignor.class semantics by shipping assignment metadata back and forth between broker and client would reintroduce the complexity and rebalance fragility that KIP-1071 was designed to eliminate (large heartbeats, client-side state, split-brain between client and broker views of the assignment). Broker-side pluggability is the natural extension point in the new protocol.