Kafka: Cannot Perform Operation After Producer Closed

You have a KafkaProducer (or Spring KafkaTemplate) that worked fine for a while, and then every send() starts throwing:

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:

org.apache.kafka.common.KafkaException: Producer closed while send in progress

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():

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:

# 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

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:

// 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<String, String> 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:

// spring-kafka 3.x — DO NOT DO THIS
@Service
class OrderPublisher {
    private final KafkaTemplate<String, String> template;

    OrderPublisher(KafkaTemplate<String, String> template) {
        this.template = template;
    }

    void publish(String key, String payload) {
        Producer<String, String> 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.

- void publish(String key, String payload) {
-     Producer<String, String> 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<String, String> 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.

- void publish(String key, String payload) {
-     Producer<String, String> 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:

- 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:

@Configuration
class KafkaProducerConfig {

    @Bean
    ProducerFactory<String, String> producerFactory() {
        Map<String, Object> 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<String, String> kafkaTemplate(ProducerFactory<String, String> 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.