apache/kafka · error · IllegalStateException

Assignor supporting the COOPERATIVE protocol violates its re

Error message

Assignor supporting the COOPERATIVE protocol violates its requirements

What it means

Thrown as an IllegalStateException from ConsumerCoordinator.validateCooperativeAssignment when a COOPERATIVE-protocol assignor returns an assignment in which a partition still owned by one member is simultaneously assigned to a different member. The cooperative protocol requires a two-step revoke-then-reassign: a partition must first be removed from its current owner's assignment and only in a later rebalance given to a new owner. Reassigning a still-owned partition directly would break the cooperative invariant and cause duplicate consumption / state corruption, so the leader rejects the assignment.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:749

            final Assignment assignment = entry.getValue();
            final Set<TopicPartition> addedPartitions = new HashSet<>(assignment.partitions());
            addedPartitions.removeAll(ownedPartitions.get(entry.getKey()));
            final Set<TopicPartition> revokedPartitions = new HashSet<>(ownedPartitions.get(entry.getKey()));
            revokedPartitions.removeAll(assignment.partitions());

            totalAddedPartitions.addAll(addedPartitions);
            totalRevokedPartitions.addAll(revokedPartitions);
        }

        // if there are overlap between revoked partitions and added partitions, it means some partitions
        // immediately gets re-assigned to another member while it is still claimed by some member
        totalAddedPartitions.retainAll(totalRevokedPartitions);
        if (!totalAddedPartitions.isEmpty()) {
            log.error("With the COOPERATIVE protocol, owned partitions cannot be " +
                "reassigned to other members; however the assignor has reassigned partitions {} which are still owned " +
                "by some members", totalAddedPartitions);

            throw new IllegalStateException("Assignor supporting the COOPERATIVE protocol violates its requirements");
        }
    }

    @Override
    protected boolean onJoinPrepare(Timer timer, int generation, String memberId) {
        log.debug("Executing onJoinPrepare with generation {} and memberId {}", generation, memberId);
        if (joinPrepareTimer == null) {
            // We should complete onJoinPrepare before rebalanceTimeoutMs,
            // and continue to join group to avoid member got kicked out from group
            joinPrepareTimer = time.timer(rebalanceConfig.rebalanceTimeoutMs);
        } else {
            joinPrepareTimer.update();
        }

        // async commit offsets prior to rebalance if auto-commit enabled
        // and there is no in-flight offset commit request
        if (autoCommitEnabled && autoCommitOffsetRequestFuture == null) {
            maybeMarkPartitionsPendingRevocation();

View on GitHub (pinned to c31c9215e1)

Solutions

Use the built-in CooperativeStickyAssignor unless you have a strong reason for a custom one (it is skipped from this validation, see line 697).
In a custom cooperative assignor, always subtract each member's currently owned partitions before reassigning them; revoke first, reassign in the next generation.
Add unit tests asserting that returned added-partitions and revoked-partitions sets are disjoint across all members.
If you cannot fix the assignor immediately, switch the group back to an eager protocol (RangeAssignor) which does not impose this constraint.

Example fix

// before (custom cooperative assignor)
@Override
public GroupAssignment assign(Cluster cluster, GroupSubscription subs) {
    // ignores each member's ownedPartitions -> may reassign owned partitions
    return eagerStyleAssignment(cluster, subs);
}

// after
@Override
public GroupAssignment assign(Cluster cluster, GroupSubscription subs) {
    // never assign a partition still owned by another member this generation;
    // leave it unassigned so it is revoked first, then reassign next round.
    Set<TopicPartition> allOwned = subs.groupSubscription().values().stream()
        .flatMap(s -> s.ownedPartitions().stream())
        .collect(Collectors.toSet());
    Map<String, Assignment> result = new HashMap<>();
    // ... compute assignment, then drop any partition in allOwned from non-owners
    return new GroupAssignment(result);
}

Defensive patterns

Strategy: validation

Validate before calling

// For any custom ConsumerPartitionAssignor used with COOPERATIVE protocol, add a unit test
// proving assign() never returns a partition that is still in a member's ownedPartitions
// in the same generation. If you cannot prove it, switch to the built-in
// org.apache.kafka.clients.consumer.CooperativeStickyAssignor.

Try / catch

try {
    consumer.poll(Duration.ofMillis(1000));
} catch (IllegalStateException e) {
    if (e.getMessage() != null && e.getMessage().contains("COOPERATIVE protocol violates")) {
        // custom cooperative assignor is buggy; fall back to CooperativeStickyAssignor
    } else {
        throw e;
    }
}

Prevention

Prefer the built-in CooperativeStickyAssignor over a custom cooperative assignor
If custom, write a regression test asserting assign() never revokes and reassigns the same partition in one generation
Never mix cooperative and eager assignors in the same consumer group
Pin the no-overlap invariant as an automated check in CI

When it happens

Trigger: A custom ConsumerPartitionAssignor advertising SupportedProtocol.COOPERATIVE (or the COOPERATIVE protocol is selected and the assignor is not the built-in CooperativeStickyAssignor) returns an assignment where the set of 'added' partitions intersects the set of 'revoked' partitions across members. The leader logs the offending partitions at line 745-747 and throws at line 749 during onLeaderElected.

Common situations: Custom cooperative assignor implemented incorrectly (treating cooperative like eager); porting an eager assignor and just tagging it COOPERATIVE without honoring incremental revocation; assignor that ignores each member's ownedPartitions when computing the new assignment; third-party assignor with a cooperative-protocol bug.

Related errors

AI-assisted analysis of apache/kafka@c31c9215e1 (2026-08-03). Data as JSON: /data/errors/823b0e110f65df96.json. Report an issue: GitHub.