apache/kafka · error · ConcurrentModificationException

KafkaShareConsumer is not safe for multi-threaded access. cu

Error message

KafkaShareConsumer is not safe for multi-threaded access. currentThread(name: ${thread.getName()}, id: ${threadId}) otherThread(id: ${currentThread.get()})

What it means

Thrown by acquire() (line 1148) as a ConcurrentModificationException when the calling thread's id differs from the thread that currently holds the consumer's single-thread lock. KafkaShareConsumer (like the classic KafkaConsumer) is not thread-safe; acquire() emulates ownership via an AtomicLong holding the owning thread id. Re-entrancy from the same thread is allowed, but a second thread touching the consumer while another is inside an API call is rejected.

Source

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

            metadata.maybeThrowBootstrapFatalException();
        } catch (RuntimeException e) {
            release();
            throw e;
        }
    }

    /**
     * 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);
    }

View on GitHub (pinned to c31c9215e1)

Solutions

Pin the consumer to a single thread — run poll, acknowledge, commit, and close on the same thread (typically a dedicated consumer thread).
If multiple threads must interact with the consumer, route all calls through a single-threaded executor (e.g. Executors.newSingleThreadExecutor) and have other threads submit tasks to it.
Use consumer.wakeup() (the only thread-safe method) to break a poll loop from another thread, then perform state changes on the consumer thread.
If you need parallel consumption, create one KafkaShareConsumer per thread, each with its own group membership or assignment.

Example fix

// before
ExecutorService pool = Executors.newFixedThreadPool(4);
pool.submit(() -> consumer.poll(Duration.ofMillis(500)));
pool.submit(() -> consumer.acknowledge(record)); // throws ConcurrentModificationException

// after
ExecutorService consumerThread = Executors.newSingleThreadExecutor();
Future<ConsumerRecords<String,String>> f = consumerThread.submit(() -> consumer.poll(Duration.ofMillis(500)));
consumerThread.submit(() -> consumer.acknowledge(record));

Defensive patterns

Strategy: validation

Validate before calling

// Pin all consumer operations to a single owning thread; cross-thread callers go through a queue.
private final java.util.concurrent.ExecutorService consumerThread =
    java.util.concurrent.Executors.newSingleThreadExecutor(r -> {
        Thread t = new Thread(r, "share-consumer-owner");
        t.setDaemon(true);
        return t;
    });

private <T> java.util.concurrent.CompletableFuture<T> submitOnOwnerThread(java.util.function.Supplier<T> task) {
    return java.util.concurrent.CompletableFuture.supplyAsync(task, consumerThread);
}

Type guard

null

Try / catch

try {
    consumer.poll(java.time.Duration.ofMillis(100));
} catch (java.util.ConcurrentModificationException e) {
    // Detected multi-threaded access: route the call through the owning thread instead of retrying in-place.
    throw new IllegalStateException("ShareConsumer accessed from a non-owner thread; serialize calls", e);
}

Prevention

A ShareConsumer is single-threaded by contract; never share the instance across threads.
To interrupt a poll from another thread, call consumer.wakeup() (which is thread-safe) instead of touching the consumer directly.
If you need fan-out, hand records off to a worker pool but keep all consumer API calls on the owner thread.

When it happens

Trigger: Thread A is inside consumer.poll(...) while thread B calls consumer.acknowledge(...), consumer.subscribe(...), consumer.commitSync(...), or any other public method. Also triggered by wakeup() paths that accidentally call into synchronized logic, or by sharing the consumer instance across an executor/pool without serialization.

Common situations: Wrapping the consumer in a Spring @Service used concurrently by multiple request threads; passing the consumer to a listener that runs on a different thread than the poll loop; scheduling poll and acknowledge on two separate scheduled executors; reusing a consumer field from both a request handler and a background rebalance listener.

Related errors

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