Fix KafkaConsumer Is Not Safe for Multi-Threaded Access

1. The Error

Exception in thread "pool-1-thread-3" java.util.ConcurrentModificationException:
KafkaConsumer is not safe for multi-threaded access.
currentThread(name: pool-1-thread-3, id: 27) otherThread(id: 1)
	at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.acquire(ClassicKafkaConsumer.java:1832)
	at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.acquireAndEnsureOpen(ClassicKafkaConsumer.java:1816)
	at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.commitSync(ClassicKafkaConsumer.java:707)
	at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1149)
	at com.example.OrderProcessor.lambda$process$0(OrderProcessor.java:52)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)

On kafka-clients 2.7 and earlier the frame is KafkaConsumer.acquire directly (the internals.ClassicKafkaConsumer split landed in 3.7 with KIP-848 groundwork). Before 2.8 the message also lacks the currentThread(...) otherThread(...) suffix, which makes older versions much harder to debug — the thread IDs were added precisely because everyone hits this.

Despite the class name, this has nothing to do with java.util collections being modified during iteration. Kafka reuses ConcurrentModificationException for its own thread-ownership guard.

Typical setup: kafka-clients 3.x (any transport — Docker local, k8s, MSK, Confluent Cloud; the broker is irrelevant, this is 100% client-side), raw KafkaConsumer with a thread pool for record processing, or a shutdown hook calling consumer.close().

The same guard fires from other entry points depending on which method you called from the wrong thread: poll(), commitAsync(), seek(), position(), pause(), resume(), close() — every public method except wakeup().

2. How to Reproduce It (step-by-step)

Single-broker KRaft cluster:

# docker-compose.yml
services:
  kafka:
    image: apache/kafka:3.8.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
      KAFKA_LISTENERS: PLAINTEXT://:29092,CONTROLLER://:29093,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

docker compose up -d
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:29092 --create --topic orders --partitions 3
docker compose exec kafka /opt/kafka/bin/kafka-producer-perf-test.sh \
  --topic orders --num-records 1000 --record-size 200 --throughput -1 \
  --producer-props bootstrap.servers=localhost:29092

The classic mistake — poll on the main thread, process and commit on a worker pool:

// kafka-clients 3.8.0
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-processor");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
ExecutorService pool = Executors.newFixedThreadPool(4);
consumer.subscribe(List.of("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> rec : records) {
        pool.submit(() -> {
            process(rec);
            // BOOM: commit from a worker thread
            consumer.commitSync(Map.of(
                new TopicPartition(rec.topic(), rec.partition()),
                new OffsetAndMetadata(rec.offset() + 1)));
        });
    }
}

Run it against the seeded topic and the exception fires within the first batch — no load, no timing luck required. If it seems intermittent in your code, it's because the guard is a check against concurrent entry: it only throws while the poll thread is simultaneously inside a consumer method. A worker-thread commit that lands while the poll thread is parked between poll() calls can slip through silently — which is worse, because now you have an unsynchronized data race that "works" in dev.

Other common triggers:

• A JVM shutdown hook or SIGTERM handler calling consumer.close() while the poll loop is blocked in poll().
• A metrics/health endpoint (Micrometer gauge, actuator) calling consumer.metrics() or consumer.position() from an HTTP thread.
• Spring Kafka: stashing the Consumer<?, ?> parameter from a @KafkaListener method into a field and touching it from a scheduler or REST controller.
• A rebalance-listener or retry framework invoking seek() from an async callback.

3. Why It Happens — Surface Level

KafkaConsumer is deliberately not thread-safe, and instead of letting you corrupt its state it fail-fasts. Every public method first calls an internal acquire(), which records the calling thread's ID. If a second thread enters while another thread is still inside the consumer, acquire() throws ConcurrentModificationException immediately.

Exactly one method is exempt and documented thread-safe: wakeup(). Everything else — poll, commitSync, commitAsync, seek, pause, close, even metrics() — must be called from the single thread that owns the consumer.

4. Why It Happens — Under the Hood

The guard is not a lock. It's a CAS-based ownership check (unchanged in spirit from 0.9 through the 3.9/4.x ClassicKafkaConsumer):

// org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer (abridged)
private static final long NO_CURRENT_THREAD = -1L;
private final AtomicLong currentThread = new AtomicLong(NO_CURRENT_THREAD);
private final AtomicInteger refCount = new AtomicInteger(0);

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(
            "KafkaConsumer is not safe for multi-threaded access. " +
            "currentThread(name: " + thread.getName() + ", id: " + threadId +
            ") otherThread(id: " + currentThread.get() + ")");
    refCount.incrementAndGet();
}

release() decrements refCount and resets currentThread to NO_CURRENT_THREAD at zero. Two consequences worth internalizing:

It detects concurrency, not foreign ownership. The consumer doesn't remember "thread 1 owns me." It only rejects a thread entering while another thread is inside. Alternating single-threaded access from different threads passes the guard — and is still broken, because the consumer's internal state (SubscriptionState, the fetch buffer, the coordinator's request queue) is plain unsynchronized mutable state with no memory-visibility guarantees across threads. The exception is the friendly failure mode. The unfriendly ones are silently lost seeks, committed offsets for records you never processed, and IllegalStateException: No current assignment for partition when one thread's view of the assignment is stale.

Why the design is single-threaded at all. Everything in the classic consumer rides on the poll thread: the fetcher's network I/O, offset commit responses, and — critically — the ConsumerRebalanceListener callbacks and JoinGroup/SyncGroup handling all execute inside poll(). Since KIP-62 (0.10.1) there is a background heartbeat thread, but it talks to the coordinator through internal synchronized structures, never through the public API — its existence is why people wrongly assume the consumer must be thread-safe. The new AsyncKafkaConsumer (group.protocol=consumer, KIP-848, GA in 4.0) moves networking to a background event loop, but the public API keeps the exact same single-threaded contract and the same acquire() guard. The threading rule is not a legacy wart that KIP-848 fixes; it's the contract.

5. The Fix

Fix 1: Commit on the poll thread (correct for most workloads)

Workers process; the poll thread owns all consumer interaction. Hand completed offsets back through a thread-safe structure:

 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
 ExecutorService pool = Executors.newFixedThreadPool(4);
+Map<TopicPartition, OffsetAndMetadata> pending = new ConcurrentHashMap<>();
 consumer.subscribe(List.of("orders"));

 while (true) {
     ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
     for (ConsumerRecord<String, String> rec : records) {
         pool.submit(() -> {
             process(rec);
-            consumer.commitSync(Map.of(
-                new TopicPartition(rec.topic(), rec.partition()),
-                new OffsetAndMetadata(rec.offset() + 1)));
+            pending.merge(
+                new TopicPartition(rec.topic(), rec.partition()),
+                new OffsetAndMetadata(rec.offset() + 1),
+                (a, b) -> b.offset() > a.offset() ? b : a);
         });
     }
+    if (!pending.isEmpty()) {
+        Map<TopicPartition, OffsetAndMetadata> snapshot = new HashMap<>(pending);
+        snapshot.forEach((tp, om) -> pending.remove(tp, om));
+        consumer.commitAsync(snapshot, (offsets, ex) -> {
+            if (ex != null) log.warn("Commit failed, will retry on next cycle", ex);
+        });
+    }
 }

Caveat you must own consciously: committing the max completed offset per partition means an out-of-order worker failure can mark earlier records as done (at-most-once for those records). If you need strict at-least-once with parallel workers, track per-partition contiguous completion (commit only up to the highest offset with no gaps below it) — or use a framework that does (see §6).

Fix 2: Shutdown — wakeup() is the only cross-thread call you get

-Runtime.getRuntime().addShutdownHook(new Thread(consumer::close));
+final AtomicBoolean running = new AtomicBoolean(true);
+Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+    running.set(false);
+    consumer.wakeup();   // thread-safe by contract; poll() throws WakeupException
+}));

 try {
-    while (true) {
+    while (running.get()) {
         ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
         ...
     }
+} catch (WakeupException e) {
+    // expected on shutdown
 } finally {
     consumer.close(Duration.ofSeconds(10));  // poll thread: leaves group cleanly
 }

Closing on the poll thread matters beyond avoiding the exception: close() sends LeaveGroup, so the partition handoff happens immediately instead of after session.timeout.ms.

Fix 3: Spring Kafka — never let the Consumer escape the listener thread

 @KafkaListener(topics = "orders")
-public void listen(ConsumerRecord<String, String> rec, Consumer<?, ?> consumer) {
-    this.stashedConsumer = consumer;   // touched later by a @Scheduled method → guard fires
+public void listen(ConsumerRecord<String, String> rec, Acknowledgment ack) {
     process(rec);
+    ack.acknowledge();                 // executed on the container's consumer thread
 }

For out-of-band operations (pause on backpressure, seek from an admin endpoint) use the container, not the consumer: registry.getListenerContainer("id").pause() is thread-safe and applied by the container's polling thread on its next cycle.

6. Best Practices & The Better Design

The design that avoids this class of bug entirely: one thread owns each consumer, full stop. Parallelism comes from either more consumers or a decoupled worker stage — never from sharing.

Option A — consumer-per-thread. Simplest correct model. N threads, each with its own KafkaConsumer, same group.id. Scaling ceiling = partition count. This is exactly what Spring Kafka's ConcurrentMessageListenerContainer gives you with concurrency=N — each of the N containers has a dedicated consumer and thread, which is why the threading problem simply never surfaces in idiomatic Spring Kafka code.

Option B — poll thread + worker pool with pause/resume backpressure. When per-record work is slow (calls to downstream services) and you'd need more parallelism than partitions:

// The poll thread is the ONLY thread touching `consumer`.
while (running.get()) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));

    if (!records.isEmpty()) {
        inFlight.addAll(submitToWorkers(records));       // workers never see the consumer
        consumer.pause(consumer.assignment());           // stop fetching, keep polling
    }

    drainCompleted(inFlight, pending);                    // collect finished offsets
    if (inFlight.isEmpty())
        consumer.resume(consumer.paused());

    commitContiguous(pending, consumer);                  // commitAsync on poll thread
}

The pause()/resume() trick is the load-bearing part: the loop keeps calling poll() so heartbeats, rebalance callbacks, and (pre-KIP-62-style) liveness via max.poll.interval.ms stay healthy while workers grind — without fetching records you can't handle yet. This is the pattern that also kills the rebalance-timeout failure mode, because poll-loop latency no longer depends on processing latency.

Option C — use something that implements Option B for you. Confluent's parallel-consumer (per-key ordering + contiguous offset tracking), Spring's @KafkaListener with AckMode.MANUAL, or Kafka Streams if the work is transformation-shaped. In 2026 there is little reason to hand-roll the offset-gap bookkeeping for a business service.

7. How to Prevent It Long-Term

• Convention: the consumer variable never leaves the method that created it. No fields, no getters, no passing into callbacks. Make this a code-review rule; it's greppable (private.*KafkaConsumer as a field is a smell outside a runnable's scope).
• Static analysis. Error Prone / ArchUnit rule forbidding KafkaConsumer references from @Scheduled, @RestController, or Runnable classes catches the stash-and-touch pattern in CI.
• Test under real concurrency. A Testcontainers integration test that runs your consumer loop while hammering your health/metrics endpoints and then sends SIGTERM will surface both the guard exception and the shutdown-hook variant before production does.
• Don't disable the pain signal. The race that slips past the guard (alternating threads) shows up as duplicate or skipped processing. Alert on committed-offset regressions and on records-consumed-rate vs downstream throughput divergence — those are the symptoms of the silent version of this bug.
• Prefer framework-managed containers. If every service hand-rolls its poll loop, every service re-solves shutdown, backpressure, and commit ordering. Standardize on Spring Kafka containers or parallel-consumer across the team and this exception disappears from your logs org-wide.

8. Key Takeaways

• ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access is Kafka's fail-fast thread-ownership guard (acquire() CAS check), not a collections bug — every public consumer method except wakeup() must run on the one thread that owns the consumer.
• The guard only fires on truly concurrent entry. Alternating access from multiple threads passes the check and silently corrupts state — the exception is the good outcome.
• Most common triggers: committing from a worker pool, consumer.close() in a shutdown hook, metrics endpoints calling consumer.metrics()/position(), and stashed Consumer references in Spring listeners.
• Fix = route every consumer call back to the poll thread: offset handoff via a concurrent map, shutdown via wakeup() + flag, Spring via Acknowledgment and container-level pause/resume.
• The single-threaded contract is unchanged in the KIP-848 AsyncKafkaConsumer — design for one-thread-per-consumer (or use parallel-consumer / Spring containers) rather than waiting for a thread-safe client that isn't coming.