Are KafkaProducer and KafkaConsumer thread-safe? How does that shape how you use each across threads?

## The core rule - **`KafkaProducer` is thread-safe** — generally you want ONE producer instance shared across all your application threads. - **`KafkaConsumer` is NOT thread-safe** — one consumer instance belongs to exactly ONE thread. ## Why the producer is shared A producer accumulates records into per-partition batches in an internal buffer (`RecordAccumulator`), and a single background thread (the *Sender*) drains batches to brokers. When many threads call `send()` on the same producer, more records land in the same batches in the same window, so batching is *better* and you amortize the connection pool and metadata. Spinning up one producer per thread wastes memory (each has its own buffer, sender thread, and connections) and produces smaller batches. So: share one producer; only make more if you genuinely need different configs (e.g. different `transactional.id`, `acks`, or compression). ## Why the consumer is single-threaded A consumer maintains a lot of mutable, non-synchronized state: the fetch buffer, current position per partition, the in-flight group membership / heartbeat state, and offset bookkeeping. Making all of that thread-safe would add locking overhead to the hot poll path. Instead the client *detects* concurrent use: before most operations it runs `acquire()` which checks that the calling thread is the one that 'owns' the consumer, and throws `ConcurrentModificationException` with the message 'KafkaConsumer is not safe for multi-threaded access' if not. Note the heartbeat is sent from *within* `poll()` (and via background coordinator logic), so you must keep calling `poll()` regularly from that one thread. ## The one cross-thread method: wakeup() `consumer.wakeup()` is the **only** method safe to call from a different thread. It sets a flag so that a blocking `poll()` returns immediately by throwing `WakeupException`. This is how you cleanly stop a poll loop from a shutdown hook or another control thread — you can't just call `close()` from the other thread, because `close()` is itself not thread-safe. ## Scaling consumption (without sharing) Kafka's scaling model is consumer *groups*, not shared instances: 1. **Thread-per-consumer**: N threads, each with its own `KafkaConsumer`, all in the same `group.id`; the group coordinator assigns partitions across them. Simple and common. 2. **Single-consumer + worker pool**: one thread polls, then hands record batches to a thread pool for processing; you must manage offset commits carefully so you don't commit records still in flight. (The detailed partition-to-thread mapping and ordering trade-offs belong to the consumer-concurrency topic; here the point is just the thread-safety contract.) ## Common mistakes - Sharing a single consumer across a thread pool 'to go faster' → `ConcurrentModificationException`. - Creating a new producer per request/message → resource waste and poor batching. - Calling `close()` from a watchdog thread to stop a poll loop → undefined/unsafe; use `wakeup()` instead.