apache/kafka · error · IllegalStateException

This RebalanceConsumer is already closed. Re-use of this obj

Error message

This RebalanceConsumer is already closed. Re-use of this object is not permitted

What it means

IllegalStateException thrown by DelegatingRebalanceConsumer.ensureOpen() when any method is called after close() set isClosed=true. The RebalanceConsumer handle (obtained via KafkaConsumer.rebalanceConsumer() / ConsumerRebalanceListener) is single-use: once closed it must not be reused. This guards against stale references after the owning consumer or the rebalance listener context has been torn down.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/DelegatingRebalanceConsumer.java:249

    public OptionalLong currentLag(TopicPartition topicPartition) {
        ensureOpen();
        return delegate.currentLag(topicPartition);
    }

    @Override
    public ConsumerGroupMetadata groupMetadata() {
        ensureOpen();
        return delegate.groupMetadata();
    }

    @Override
    public void close() {
        isClosed = true;
    }

    private void ensureOpen() {
        if (isClosed) {
            throw new IllegalStateException(
                    "This RebalanceConsumer is already closed. Re-use of this object is not permitted");
        }
    }
}

View on GitHub (pinned to c31c9215e1)

Solutions

1. Stop invoking methods on the RebalanceConsumer after close(); obtain a fresh instance from the new KafkaConsumer.rebalanceConsumer().
2. Null out cached references to the RebalanceConsumer in ConsumerRebalanceListener.onPartitionsRevoked once close() is called.
3. Create a new KafkaConsumer rather than reusing a closed one.

Example fix

// before
RebalanceConsumer rc = consumer.rebalanceConsumer();
consumer.close();
rc.subscription(); // IllegalStateException

// after
consumer.close();
// discard old reference
try (KafkaConsumer<K,V> c2 = new KafkaConsumer<>(props)) {
    c2.rebalanceConsumer().subscription(); // fresh handle
}

// RebalanceConsumer (DelegatingRebalanceConsumer) exposes no isOpen()/isClosed() getter.
// Track lifecycle yourself in the owning object and guard every call:
if (rebalanceConsumer == null || rebalanceConsumerClosed) {
    throw new IllegalStateException("RebalanceConsumer is closed; obtain a new instance");
}
rebalanceConsumer.committed(...); // safe to use

// Optional state wrapper that makes 'closed' visible in the type system.
final class OpenRebalanceConsumer {
    private final RebalanceConsumer delegate;
    private OpenRebalanceConsumer(RebalanceConsumer c) { this.delegate = c; }
    static OpenRebalanceConsumer open(RebalanceConsumer c) {
        return new OpenRebalanceConsumer(c);
    }
    RebalanceConsumer get() { return delegate; } // only callable while wrapper exists
}
// Closing invalidates the wrapper (drop the reference); a fresh open() is required to use it again.

try {
    rebalanceConsumer.poll(...);
} catch (IllegalStateException e) {
    if (e.getMessage() != null && e.getMessage().contains("already closed")) {
        // DelegatingRebalanceConsumer.ensureOpen() fired at line 249.
        // Discard this reference and obtain a new consumer from KafkaConsumer.
        log.warn("RebalanceConsumer reused after close; discarding");
        rebalanceConsumer = null; // force re-acquisition
    } else {
        throw e;
    }
}

When it happens

Trigger: Calling any method on the RebalanceConsumer object (assignment, subscription, listTopics, currentLag, groupMetadata, etc.) after its close() was invoked — for example using a cached rebalanceConsumer reference inside a ConsumerRebalanceListener after the consumer closed.

Common situations: Storing the RebalanceConsumer in a field and reusing it across consumer lifecycles; calling rebalanceConsumer inside @KafkaListener tear-down after the consumer was already closed; frameworks that pool/recreate consumers while the listener holds a stale RebalanceConsumer.

Related errors

• This consumer has already been closed.
• You can only check the position for partitions assigned to t
• The timeout cannot be negative.
• Failed to close kafka consumer
• Operation timed out before completion

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