{"record":{"id":"e556100ffed6378c","repo":"apache/pulsar","slug":"producercache-is-already-closed","errorCode":null,"errorMessage":"ProducerCache is already closed","messagePattern":"ProducerCache is already closed","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java","lineNumber":133,"sourceCode":"                                        return null;\n                                    });\n                    if (closed.get()) {\n                        closeFutures.add(closeFuture);\n                    }\n                })\n                .weigher((key, producer) -> Math.max(producer.getNumOfPartitions(), 1))\n                .maximumWeight(PRODUCER_CACHE_MAX_SIZE);\n        if (PRODUCER_CACHE_TIMEOUT_SECONDS > 0) {\n            builder.expireAfterAccess(Duration.ofSeconds(PRODUCER_CACHE_TIMEOUT_SECONDS));\n        }\n        cache = builder.build();\n        CacheMetricsCollector.CAFFEINE.addCache(\"function-producer-cache\", cache);\n    }\n\n    public <T> Producer<T> getOrCreateProducer(CacheArea cacheArea, String topicName, Object additionalCacheKey,\n                                               Callable<Producer<T>> supplier) {\n        if (closed.get()) {\n            throw new IllegalStateException(\"ProducerCache is already closed\");\n        }\n        @SuppressWarnings(\"unchecked\")\n        Producer<T> producer = (Producer<T>) cache.get(\n                new ProducerCacheKey(cacheArea, topicName, additionalCacheKey), key -> {\n            try {\n                return supplier.call();\n            } catch (RuntimeException e) {\n                throw e;\n            } catch (Exception e) {\n                throw new RuntimeException(\"Unable to create producer for topic '\" + topicName + \"'\", e);\n            }\n        });\n        return producer;\n    }\n\n    public void close() {\n        if (closed.compareAndSet(false, true)) {\n            cache.invalidateAll();","sourceCodeStart":115,"sourceCodeEnd":151,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java#L115-L151","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nproducerCache.getOrCreateProducer(cacheArea, topic, null, supplier); // may throw after close\n// after\nif (!isShuttingDown.get()) {\n    try {\n        producerCache.getOrCreateProducer(cacheArea, topic, null, supplier);\n    } catch (IllegalStateException e) {\n        // cache closed during shutdown; drop the record or buffer it\n    }\n}","handlingStrategy":"try-catch","validationCode":"// before producing, if you control the cache lifecycle\n// ProducerCache exposes no public isOpen(); track shutdown yourself\nprivate final AtomicBoolean shuttingDown = new AtomicBoolean(false);\nboolean canProduce = !shuttingDown.get();","typeGuard":null,"tryCatchPattern":"try {\n    Producer<T> p = cache.getOrCreateProducer(area, topic, key, supplier);\n} catch (IllegalStateException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"already closed\")) {\n        // benign shutdown race: drop or buffer the record\n    } else {\n        throw e;\n    }\n}","preventionTips":["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."],"tags":["lifecycle","producer-cache","illegal-state","shutdown"],"backgroundTag":"resource-already-closed","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}