Fix Pulsar Consumer Already Closed Error (2026)

The Pulsar client library is failing to send messages to a broker because the producer object, which represents your application’s connection to Pulsar, has been prematurely closed.

Common Causes and Fixes

1. Producer Closed by Client Library Due to Network Instability:

Diagnosis: Check the Pulsar client logs for ConnectionClosedException or similar network-related errors occurring around the same time as your "Consumer already closed" error. Look for messages indicating the client lost its connection to a broker.

Fix: In your client code, implement robust error handling for producer creation and message sending. Specifically, wrap your producer.sendAsync() calls in a try-catch block and if an exception occurs, attempt to re-create the producer. For Java, this might look like:

try {
    producer.sendAsync(message).thenAccept(msgId -> {
        // Handle successful send
    }).exceptionally(e -> {
        // Handle send failure, potentially re-create producer
        if (e.getCause() instanceof org.apache.pulsar.client.api.PulsarClientException.ConnectionClosedException) {
            // Re-create producer logic here
        }
        return null;
    });
} catch (Exception e) {
    // Handle producer creation failure or other immediate issues
}

Why it works: This approach allows the client library to gracefully recover from transient network issues. By catching ConnectionClosedException, you can trigger a re-initialization of the producer, establishing a new connection to the broker and allowing subsequent messages to be sent.

2. Producer Closed Explicitly in Application Logic:

Diagnosis: Search your application’s codebase for any explicit calls to producer.close() or producer.closeAsync(). Pay close attention to code paths that handle application shutdown, error conditions, or resource cleanup.
Fix: Ensure that producer.close() is only called when your application is intentionally shutting down and no longer needs to send messages. If you’re closing the producer in response to an error, reconsider that strategy and opt for re-creation as described above. If you have multiple threads or components managing the producer, ensure there’s a clear, single source of truth for its lifecycle.
Why it works: The "Consumer already closed" error arises because the producer object you’re trying to use has had its underlying connection and resources released. By preventing premature or unintended calls to close(), you keep the producer alive and functional for as long as your application requires it.

3. Producer Timeout Due to Broker Unresponsiveness:

Diagnosis: Examine Pulsar broker logs for any signs of high load, long garbage collection pauses, or unresponsiveness. Also, check client logs for TimeoutException or messages indicating the producer is being marked as inactive by the broker.

Fix: Adjust the producer’s sendTimeoutMs configuration. For example, if you’re using Java:

PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://localhost:6650")
    .build();

Producer<byte[]> producer = client.newProducer()
    .topic("my-topic")
    .sendTimeout(60, TimeUnit.SECONDS) // Increased timeout
    .create();

Why it works: If brokers are slow to acknowledge messages due to high load or internal issues, the producer might time out waiting for an acknowledgment. Increasing the sendTimeoutMs gives the broker more time to respond, preventing the producer from being prematurely closed due to perceived unresponsiveness.

4. Producer Leaked and Closed by Garbage Collector (Less Common):

Diagnosis: This is harder to diagnose directly. If you’re not explicitly closing producers but still see this error, and other causes are ruled out, it’s possible your producer objects are being garbage collected. This often happens if producers are created within short-lived scopes and not held onto by a long-lived object.
Fix: Ensure that your Producer objects are held by a long-lived object (e.g., a singleton or a managed service) for the duration of your application’s need to send messages. Avoid creating producers inside loops or methods that are called frequently if the producer is intended to be persistent.
Why it works: By keeping a strong reference to the Producer object, you prevent the Java garbage collector (or equivalent in other languages) from reclaiming its memory and resources, which would effectively "close" it from the application’s perspective.

5. Broker Restart or Failover During Operation:

Diagnosis: Correlate the timing of your error with any known Pulsar cluster maintenance, restarts, or broker failover events. Check Pulsar admin logs or monitoring for cluster activity.
Fix: Implement a reconnection strategy in your client application. When a ConnectionClosedException or similar is caught during message sending, the client should automatically attempt to reconnect to the Pulsar cluster and re-create the producer. The PulsarClient builder often has settings for automatic reconnection, but explicit handling in your exceptionally block (as shown in Cause 1) is more reliable for producer-specific recovery.
Why it works: When a broker restarts or fails over, existing connections are terminated. A well-implemented reconnection strategy ensures that the client can re-establish a connection to a healthy broker and continue its operations by recreating the producer.

6. Misconfiguration of Producer enableBatching and batchingMaxPublishDelay:

Diagnosis: If batching is enabled and misconfigured, messages might be held in the producer’s batch buffer for too long, leading to timeouts or unexpected producer state changes if the client library’s internal timers are not synchronized with the application’s perception of time.

Fix: Ensure batchingMaxPublishDelay is set to a reasonable value (e.g., 100 milliseconds) if enableBatching is true.

Producer<byte[]> producer = client.newProducer()
    .topic("my-topic")
    .enableBatching(true)
    .batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS)
    .create();

Why it works: This setting controls how long messages wait in the batch before being sent. A very long delay, combined with other system latencies, could contribute to timeouts that indirectly lead to the producer being closed. Tuning this value ensures timely sending of batched messages.

The next error you’ll likely encounter if you fix this is a TopicNotFoundException if the topic specified in your producer configuration doesn’t exist or if permissions are insufficient.