> ## Content Index > Fetch the complete content index at: https://www.ggorantala.dev/llms.txt > Use this file to discover other available public pages before exploring further. # Kafka: Cannot Perform Operation After Producer Closed - URL: https://www.ggorantala.dev/kafka-cannot-perform-operation-after-producer-closed-fix/ - Published: 2026-02-17T14:05:00.000Z - Updated: 2026-08-29T08:19:14.000Z - Description: Kafka IllegalStateException 'Cannot perform operation after producer has been closed' explained: why it happens, the Spring share - Author: Gopi Gorantala - Tags: kafka-errors, kafka-producer, IllegalStateException, spring-kafka, Java, event-streaming You have a `KafkaProducer` (or Spring `KafkaTemplate`) that worked fine for a while, and then every `send()` starts throwing: ```text java.lang.IllegalStateException: Cannot perform operation after producer has been closed at org.apache.kafka.clients.producer.KafkaProducer.throwIfProducerClosed(KafkaProducer.java:1073) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:1002) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:988) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:947) ``` Two sibling messages come out of the same lifecycle bug and belong in the same mental bucket: ```text org.apache.kafka.common.KafkaException: Producer closed while send in progress ``` ```text java.lang.IllegalStateException: Producer is closed forcefully. ``` This is not a broker problem, a network problem, or a config-value problem. It is a **client lifecycle** problem: something closed the producer, and then something else tried to use it. The whole job is figuring out *who closed it*, because on a raw `kafka-clients` producer and on Spring's shared producer the "who" is completely different. `kafka-clients` 2.x–4.0, Spring for Apache Kafka 2.x–4.0, KRaft or ZooKeeper — the mechanics below are broker-version-independent. ## 1\. The Error The exact throw site is `KafkaProducer.throwIfProducerClosed()`, called at the top of `doSend()`: ```java private void throwIfProducerClosed() { if (sender == null || !sender.isRunning()) throw new IllegalStateException("Cannot perform operation after producer has been closed"); } ``` The `Sender` is the background I/O thread (`kafka-producer-network-thread`). `close()` stops it. Once stopped, `sender.isRunning()` is `false` forever — the producer object is a one-way latch. There is no `reopen()`. Any `send()`, `partitionsFor()`, `beginTransaction()`, `flush()`, or `abortTransaction()` after close throws immediately, on the calling thread, before anything hits the accumulator. The `Producer closed while send in progress` variant is thrown when `close()` fires while another thread is mid-`send()`; the in-flight call loses the race. `Producer is closed forcefully` appears when `close(Duration)` times out and force-closes with records still pending — those records' callbacks complete exceptionally with this cause. Where it typically bites: - Spring Boot services where someone manually `close()`d the `KafkaTemplate`'s producer or called `producerFactory.reset()` mid-traffic. - POC code that wraps a producer in try-with-resources inside a method that gets called more than once, or shares the producer with another thread. - A `Callback` or error handler that closes the producer on the first failure, while other threads keep sending. - Application shutdown ordering: a `@PreDestroy` or shutdown hook closes the producer before the last senders drain. ## 2\. How to Reproduce It Minimal single-broker KRaft cluster: ```yaml # docker-compose.yml services: kafka: image: apache/kafka:3.9.0 ports: - "9092:9092" environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 ``` ```bash docker compose up -d docker exec -it $(docker ps -qf name=kafka) \ /opt/kafka/bin/kafka-topics.sh --create --topic orders \ --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1 ``` The raw-client reproduction — reuse a producer after closing it. This is the try-with-resources trap in loop form: ```java // kafka-clients 3.9.0 — DO NOT DO THIS Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("key.serializer", StringSerializer.class.getName()); props.put("value.serializer", StringSerializer.class.getName()); Producer producer = new KafkaProducer<>(props); producer.send(new ProducerRecord<>("orders", "k1", "first")); producer.close(); // network thread stops here producer.send(new ProducerRecord<>("orders", "k2", "second")); // -> IllegalStateException: Cannot perform operation after producer has been closed ``` The Spring reproduction, which is the one people actually hit in services. The `KafkaTemplate` hands you the *shared* producer; if you close it, you've closed it for the whole application: ```java // spring-kafka 3.x — DO NOT DO THIS @Service class OrderPublisher { private final KafkaTemplate template; OrderPublisher(KafkaTemplate template) { this.template = template; } void publish(String key, String payload) { Producer p = template.getProducerFactory().createProducer(); try { p.send(new ProducerRecord<>("orders", key, payload)); } finally { p.close(); // closes the SHARED singleton for the whole app } } } ``` The first call succeeds. Every subsequent call — from this bean or any other — throws `Cannot perform operation after producer has been closed`, because `createProducer()` returns the *same* underlying producer each time. ## 3\. Why It Happens — Surface Level A `KafkaProducer` is a heavyweight, thread-safe object designed to be created once and shared across threads for the life of the application. `close()` is terminal. The error means your code has an ordering bug: a close happened, then a use happened, on an object that cannot be revived. On the raw client it is usually a scope/loop mistake. In Spring it is almost always that application code reached past `KafkaTemplate` and closed a producer it does not own. ## 4\. Why It Happens — Under the Hood **Raw client.** The constructor spins up the `Sender` runnable on a dedicated thread and a `RecordAccumulator` buffer. `send()` serializes the record, appends it to a per-partition batch in the accumulator, and returns a `Future`; the `Sender` drains batches to brokers asynchronously. `close()` sets the sender's running flag false, flushes (or force-aborts on timeout), joins the I/O thread, and closes the network client and metrics. After that, `sender.isRunning()` is permanently false. `throwIfProducerClosed()` is the guard that turns "use after close" into a clean exception instead of a `NullPointerException` deep in the accumulator. **Spring's shared-producer model** is where most of the confusion lives, so it is worth being precise. `DefaultKafkaProducerFactory` is a `ProducerFactory` that, for non-transactional configs, holds **one singleton `KafkaProducer`** and returns it from every `createProducer()` call. Critically, the object it returns is a **wrapper** (`CloseSafeProducer`). Calling `close()` on that wrapper is a **no-op against the physical producer** — it exists precisely so that library code and try-with-resources can "close" a producer without destroying the shared instance. So how do you end up with a physically closed producer? Three ways: 1. You call `factory.reset()` (or the actuator/`ContextStoppedEvent` path) which *physically* closes the singleton and clears the transactional cache. If traffic is still flowing, the next `send()` on a reference obtained *before* the reset hits a closed delegate. 2. You unwrap or hold a reference to the real producer and call the real `close()`. 3. Application shutdown publishes `ContextStoppedEvent` / runs `DisposableBean.destroy()`, physically closing the producer while a late thread still sends. The factory closes on `destroy()` or `ContextStoppedEvent` by design. For **transactional** producers the factory keeps a **cache/pool** keyed by `transactional.id` suffix, one producer per concurrent transaction. Here the historical footgun (seen in third-party pooling layers built on top, e.g. the Axon Kafka extension bug) is that a producer gets physically closed but still handed back out of the cache, so the *next* transaction opens on a dead delegate and throws `Cannot perform operation after producer has been closed`. The lesson generalizes: **let the factory own physical lifecycle; never close what you got from `createProducer()` yourself.** **Per-thread producers** (`producerPerThread = true`) invert the rule: those are *not* closed by `destroy()` or `reset()`, and the caller **must** call `closeThreadBoundProducer()` to release them. Mixing that mental model up — closing a shared producer as if it were thread-bound, or leaking a thread-bound one — produces the same exception or a slow file-descriptor leak. ## 5\. The Fix **Raw client — create once, share, close once at shutdown.** ```diff - void publish(String key, String payload) { - Producer producer = new KafkaProducer<>(props); - try { - producer.send(new ProducerRecord<>("orders", key, payload)); - } finally { - producer.close(); // new producer per call, closed each time - } - } + // one instance for the app; injected/singleton, thread-safe + private final Producer producer = new KafkaProducer<>(props); + + void publish(String key, String payload) { + producer.send(new ProducerRecord<>("orders", key, payload)); + } + + @PreDestroy // or a shutdown hook — the ONLY place close() belongs + void shutdown() { + producer.close(Duration.ofSeconds(30)); + } ``` **Spring — never close the factory's producer; just use `KafkaTemplate`.** ```diff - void publish(String key, String payload) { - Producer p = template.getProducerFactory().createProducer(); - try { - p.send(new ProducerRecord<>("orders", key, payload)); - } finally { - p.close(); // kills the shared singleton - } - } + void publish(String key, String payload) { + template.send("orders", key, payload); // template manages the producer + } ``` If you genuinely need a raw `Producer` from the factory (interop code), obtain it, use it, and call `close()` on the returned wrapper — it is a safe no-op — but **do not** reach for the underlying instance and do not call `reset()` on a live path. **Timeout variant (`Producer is closed forcefully`).** If shutdown force-closes with pending records, give `close()` a real budget and flush first: ```diff - producer.close(); // default: waits Long.MAX_VALUE, or forced on interrupt + producer.flush(); // block until buffered records are acked + producer.close(Duration.ofSeconds(30)); // bounded, graceful drain ``` Which fix applies depends on the cause: reused/looped producer → hoist to a singleton; Spring app-code closing the shared producer → delete the `close()` and use `KafkaTemplate`; shutdown-ordering race → make the producer the last thing closed, after listeners/executors stop. ## 6\. Best Practices & The Better Design The producer is infrastructure, not a per-request object. Model it that way: ```java @Configuration class KafkaProducerConfig { @Bean ProducerFactory producerFactory() { Map cfg = new HashMap<>(); cfg.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); cfg.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); cfg.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); cfg.put(ProducerConfig.ACKS_CONFIG, "all"); cfg.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // default since 3.0 return new DefaultKafkaProducerFactory<>(cfg); } @Bean KafkaTemplate kafkaTemplate(ProducerFactory pf) { return new KafkaTemplate<>(pf); // ONE shared, thread-safe producer, Spring owns close } } ``` Application code then only ever touches `KafkaTemplate`. It is fully thread-safe; a single instance serves your entire request-handling thread pool. There is no scenario in normal service code where you should be calling `createProducer()` or `close()` yourself. Design rules that make this class of bug structurally impossible: - **One producer per app (or per pool the framework owns).** Inject it; never `new KafkaProducer<>()` inside a request or loop. - **Close is a shutdown concern.** The only legitimate `close()` lives in `@PreDestroy`, a shutdown hook, or the framework's own lifecycle — never in business logic, never in a `finally` around a single send. - **Don't catch a send failure by closing the producer.** A failed `send()` does not require a new producer; with `enable.idempotence=true` (default) the client already retries safely. Closing on error is what turns one bad record into an app-wide outage. - **For transactions, let Spring drive it** via `transaction-id-prefix` and `executeInTransaction` / `@Transactional`, so the factory's cache — not your code — governs producer physical lifecycle. ## 7\. How to Prevent It Long-Term - **Lint / ArchUnit rule:** ban `close()` calls on any `Producer`/`KafkaProducer` reference outside classes annotated for lifecycle (config/shutdown). Ban `getProducerFactory().createProducer()` in service packages. - **Ban `new KafkaProducer<>` outside a single factory/config class** so producer creation is centralized and auditable. - **Shutdown ordering test:** an integration test that sends, triggers context close, and asserts no `Cannot perform operation after producer has been closed` surfaces on in-flight sends. Testcontainers makes this cheap — start a broker, publish under a small executor, close the context, join the executor, assert clean drain. - **Alert on the exception itself.** Unlike retriable broker errors, this is always a code/lifecycle defect — any occurrence in production is signal, not noise. Wire it to a log-based alert. - **Monitor `record-error-rate` and `record-send-rate`.** A cliff in send-rate to zero with a spike in error-rate right after a deploy or actuator call is the fingerprint of a producer that got closed under live traffic. ## 8\. Key Takeaways - `Cannot perform operation after producer has been closed` is a **use-after-close** lifecycle bug — `KafkaProducer.close()` is terminal, the object never reopens. - **A `KafkaProducer` is meant to be created once and shared across threads.** Per-request creation, try-with-resources in a loop, or closing on error all cause this. - In **Spring**, `KafkaTemplate` / `DefaultKafkaProducerFactory` hand you a *shared singleton wrapped* producer; closing what `createProducer()` returns is a no-op, but `reset()` / context-stop / grabbing the real delegate physically closes it for the whole app. - **The fix is almost always deletion:** remove the stray `close()`, inject one producer/`KafkaTemplate`, and let the framework own physical shutdown. - **Close belongs only in shutdown**, bounded with `close(Duration)` after `flush()`; a send failure is never a reason to close the producer.