apache/kafka · error · IllegalStateException

KafkaShareConsumer methods are not accessible from user-defi

Error message

KafkaShareConsumer methods are not accessible from user-defined acknowledgement commit callback.

What it means

Thrown by acquire() (line 1153) when acknowledgementCommitCallbackHandler.hasEnteredCallback() is true. It prevents re-entrancy: while the consumer is invoking the user-supplied acknowledgement commit callback, calling any ShareConsumer API method from inside that callback would corrupt in-flight acknowledgement state. The guard short-circuits such calls with an IllegalStateException.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:1153

    }

    /**
     * Acquire the light lock protecting this consumer from multithreaded access. Instead of blocking
     * when the lock is not available, however, we just throw an exception (since multithreaded usage is not
     * supported).
     *
     * @throws ConcurrentModificationException if another thread already has the lock
     */
    private void acquire() {
        final Thread thread = Thread.currentThread();
        final long threadId = thread.getId();
        if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId))
            throw new ConcurrentModificationException("KafkaShareConsumer is not safe for multi-threaded access. " +
                    "currentThread(name: " + thread.getName() + ", id: " + threadId + ")" +
                    " otherThread(id: " + currentThread.get() + ")"
            );
        if (acknowledgementCommitCallbackHandler != null && acknowledgementCommitCallbackHandler.hasEnteredCallback()) {
            throw new IllegalStateException("KafkaShareConsumer methods are not accessible from user-defined " +
                    "acknowledgement commit callback.");
        }
        refCount.incrementAndGet();
    }

    /**
     * Release the light lock protecting the consumer from multithreaded access.
     */
    private void release() {
        if (refCount.decrementAndGet() == 0)
            currentThread.set(NO_CURRENT_THREAD);
    }

    public static LogContext createLogContext(final String clientId, final String groupId) {
        return new LogContext("[ShareConsumer clientId=" + clientId + ", groupId=" + groupId + "] ");
    }

    private void maybeThrowInvalidGroupIdException() {

View on GitHub (pinned to c31c9215e1)

Solutions

1. Do not call any ShareConsumer method from inside the AcknowledgementCommitCallback — only inspect the completed-acknowledgements data you are given.
2. If follow-up work is needed, hand it off to a separate executor or set a flag that the main poll loop checks.
3. Use the callback only for side effects that do not touch the consumer: external metrics, logging, business notifications.
4. Move the logic you attempted in the callback into the main consumer loop, executed after poll() returns.

Example fix

// before
consumer.setAcknowledgementCommitCallback(acks -> {
    consumer.commitSync(); // throws IllegalStateException
});

// after
consumer.setAcknowledgementCommitCallback(acks -> {
    metrics.counter("acks.committed").increment(acks.size());
});

// Keep the acknowledgement commit callback pure: do not call consumer methods inside it.
org.apache.kafka.clients.consumer.AcknowledgementCommitCallback cb = (acknowledgements, error) -> {
    // Allowed: log, persist offsets to an external store, emit metrics.
    // Forbidden here: consumer.poll(...), consumer.acknowledge(...), consumer.close(), etc.
    if (error != null) log.warn("Ack commit failed", error);
};
consumer.setAcknowledgementCommitCallback(cb);

null

// Defensive guard inside any helper that might be invoked from the callback.
if (Thread.currentThread().getName().contains("ack-commit")) {
    throw new IllegalStateException("Refusing to call consumer methods from acknowledgement commit callback");
}

When it happens

Trigger: Inside a custom AcknowledgementCommitCallback, the code calls consumer.acknowledge(...), consumer.commitSync(...), consumer.poll(...), or any other public ShareConsumer method. Because the callback is dispatched on the consumer thread during handleCompletedAcknowledgements, the re-entrant acquire() detects the callback is in progress and throws.

Common situations: Developers treating the commit callback like an event listener and trying to drive further consumption from it (e.g. polling for more records or acknowledging again); porting code from a different MQ API where callback-driven acknowledgement chaining is idiomatic; logging/metrics hooks that accidentally call consumer.metrics() or other API methods.

Related errors

• All records must be acknowledged in explicit acknowledgement
• Implicit acknowledgement of delivery is being used.
• The record cannot be acknowledged.
• The record cannot be acknowledged.
• Unknown acknowledge type id: {id}

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