What causes "ProducerCache is already closed" in apache/pulsar?

Ensure the function instance stops producing before shutdown: complete or cancel in-flight processing before ProducerCache.close() is called.

apache/pulsar · error · IllegalStateException

ProducerCache is already closed

Error message

ProducerCache is already closed

What it means

ProducerCache.getOrCreateProducer throws IllegalStateException when the cache has already been closed via close(). The cache guards with an AtomicBoolean `closed` so that no new producers are created after instance shutdown begins, preventing leaks of Pulsar producers that would never be cleaned up. If you see this, code is still trying to produce after the function instance is shutting down.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java:133

                                        return null;
                                    });
                    if (closed.get()) {
                        closeFutures.add(closeFuture);
                    }
                })
                .weigher((key, producer) -> Math.max(producer.getNumOfPartitions(), 1))
                .maximumWeight(PRODUCER_CACHE_MAX_SIZE);
        if (PRODUCER_CACHE_TIMEOUT_SECONDS > 0) {
            builder.expireAfterAccess(Duration.ofSeconds(PRODUCER_CACHE_TIMEOUT_SECONDS));
        }
        cache = builder.build();
        CacheMetricsCollector.CAFFEINE.addCache("function-producer-cache", cache);
    }

    public <T> Producer<T> getOrCreateProducer(CacheArea cacheArea, String topicName, Object additionalCacheKey,
                                               Callable<Producer<T>> supplier) {
        if (closed.get()) {
            throw new IllegalStateException("ProducerCache is already closed");
        }
        @SuppressWarnings("unchecked")
        Producer<T> producer = (Producer<T>) cache.get(
                new ProducerCacheKey(cacheArea, topicName, additionalCacheKey), key -> {
            try {
                return supplier.call();
            } catch (RuntimeException e) {
                throw e;
            } catch (Exception e) {
                throw new RuntimeException("Unable to create producer for topic '" + topicName + "'", e);
            }
        });
        return producer;
    }

    public void close() {
        if (closed.compareAndSet(false, true)) {
            cache.invalidateAll();

View on GitHub (pinned to 820761864e)

Solutions

Ensure the function instance stops producing before shutdown: complete or cancel in-flight processing before ProducerCache.close() is called.
Check the calling code's lifecycle — do not hold references to ProducerCache beyond instance shutdown (e.g. in executor threads or async callbacks).
In tests, make sure all producer work completes before closing the cache (join futures before @After teardown).
Catch IllegalStateException and treat it as a benign shutdown race if your component is expected to be torn down concurrently.

Example fix

// before
producerCache.getOrCreateProducer(cacheArea, topic, null, supplier); // may throw after close
// after
if (!isShuttingDown.get()) {
    try {
        producerCache.getOrCreateProducer(cacheArea, topic, null, supplier);
    } catch (IllegalStateException e) {
        // cache closed during shutdown; drop the record or buffer it
    }
}

Defensive patterns

Strategy: try-catch

Validate before calling

// before producing, if you control the cache lifecycle
// ProducerCache exposes no public isOpen(); track shutdown yourself
private final AtomicBoolean shuttingDown = new AtomicBoolean(false);
boolean canProduce = !shuttingDown.get();

Try / catch

try {
    Producer<T> p = cache.getOrCreateProducer(area, topic, key, supplier);
} catch (IllegalStateException e) {
    if (e.getMessage() != null && e.getMessage().contains("already closed")) {
        // benign shutdown race: drop or buffer the record
    } else {
        throw e;
    }
}

Prevention

Never cache or share ProducerCache beyond the function instance lifecycle.
Complete all async producer work before calling close().
In tests, join outstanding futures before teardown methods close the cache.

When it happens

Trigger: Calling getOrCreateProducer (directly or via instance output producer creation) after ProducerCache.close() has been invoked — e.g. a function's process call racing with instance shutdown, or calling the cache from a thread that outlives the instance lifecycle.

Common situations: Function termination/timeout while a batch of messages is still being processed; async callbacks completing after the runtime closes the cache; tests that close the cache in @AfterEach while an async task still runs.

Related errors

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/e556100ffed6378c. Report an issue: GitHub.