apache/kafka · error · IllegalStateException

Cannot lose partitions that are not currently assigned: {not

Error message

Cannot lose partitions that are not currently assigned: {notAssigned}

What it means

Thrown by MockConsumer.losePartitions when one or more of the requested partitions is not in the current assignment. MockConsumer simulates rebalance events for tests; losing a partition that was never assigned would put the mock in an inconsistent state, so the call is rejected wholesale. The message lists exactly which partitions are not assigned.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java:172

     * Simulates a partition loss event. Calls {@link ConsumerRebalanceListener#onPartitionsLost}
     * for the specified partitions and removes them from the current assignment. Unlike
     * {@link #rebalance(Collection)}, which calls {@link ConsumerRebalanceListener#onPartitionsRevoked},
     * this method models the case where the consumer loses partitions without a graceful revoke..
     *
     * <p>Only records belonging to the lost partitions are cleared; records for retained
     * partitions are unaffected.
     *
     * @param partitionsLost the partitions to lose; all must be currently assigned
     * @throws IllegalStateException if any partition is not currently assigned
     */
    public synchronized void losePartitions(Collection<TopicPartition> partitionsLost) {
        Set<TopicPartition> currentAssignment = this.subscriptions.assignedPartitions();
        Set<TopicPartition> lost = new HashSet<>(partitionsLost);
        List<TopicPartition> notAssigned = lost.stream()
            .filter(tp -> !currentAssignment.contains(tp))
            .collect(Collectors.toList());
        if (!notAssigned.isEmpty())
            throw new IllegalStateException("Cannot lose partitions that are not currently assigned: " + notAssigned);
        lost.forEach(records::remove);
        this.subscriptions.onPartitionsLost(lost);
        Set<TopicPartition> remaining = currentAssignment.stream()
            .filter(tp -> !lost.contains(tp))
            .collect(Collectors.toSet());
        this.subscriptions.assignFromSubscribed(remaining);
    }

    @Override
    public synchronized Set<String> subscription() {
        return subscriptions.subscription();
    }

    @Override
    public synchronized void subscribe(Collection<String> topics) {
        subscribeInternal(topics, null);
    }

View on GitHub (pinned to 996fb4585a)

Solutions

1. Verify partitionsLost is a subset of mock.subscription()/assignedPartitions() before calling losePartits; assert in the test setup.
2. Call mock.assign(...) (or schedule rebalance) first so the partitions are in the assignment.
3. Drop the offending TopicPartition from the lost set or re-derive it from the current assignment.

Example fix

// before
mockConsumer.assign(Set.of(new TopicPartition("orders", 0)));
mockConsumer.losePartitions(Set.of(new TopicPartition("orders", 1))); // not assigned

// after
TopicPartition tp0 = new TopicPartition("orders", 0);
mockConsumer.assign(Set.of(tp0));
mockConsumer.losePartitions(Set.of(tp0));

Set<TopicPartition> assigned = mockConsumer.assignment();
List<TopicPartition> invalid = partitionsLost.stream()
    .filter(tp -> !assigned.contains(tp)).collect(Collectors.toList());
if (!invalid.isEmpty()) throw new IllegalStateException("Cannot lose unassigned: " + invalid);
mockConsumer.losePartitions(partitionsLost);

static boolean allAssigned(MockConsumer<?,?> m, Collection<TopicPartition> tps) {
    Set<TopicPartition> a = m.assignment();
    return tps.stream().allMatch(a::contains);
}

try {
    mockConsumer.losePartitions(toLose);
} catch (IllegalStateException e) {
    if (e.getMessage() != null && e.getMessage().startsWith("Cannot lose partitions that are not currently assigned")) {
        // recompute toLose as the intersection with current assignment, then retry once
    }
    throw e;
}

When it happens

Trigger: In a unit test, calling losePartitions with a TopicPartition you never assigned via assign() or rebalance; calling losePartitions twice for the same partition; mismatched partition numbers between assign and lose.

Common situations: Test scaffolding where the assigned set and the lost set are computed by different helpers; refactor that changed partition counts; copy-paste of a test that used a different topic name.

Related errors

• Cannot add records for a partition that is not assigned to t
• You can only check the position for partitions assigned to t
• The partition {} does not have a beginning offset.
• The partition {} does not have an end offset.
• This consumer has already been closed.

AI-assisted analysis of apache/kafka@996fb4585a (2026-08-11). Data as JSON: /api/errors/1becbb88ab3b6df2. Report an issue: GitHub.