{"record":{"id":"3853513391d7f768","repo":"apache/flink","slug":"failed-to-acquire-d-out-of-d-permits-to-send-val","errorCode":null,"errorMessage":"Failed to acquire %d out of %d permits to send value in %s.","messagePattern":"Failed to acquire (.+?) out of (.+?) permits to send value in (.+?)\\.","errorType":"exception","errorClass":"TimeoutException","httpStatus":null,"severity":"warning","filePath":"flink-core/src/main/java/org/apache/flink/api/common/io/SinkUtils.java","lineNumber":52,"sourceCode":"     * Acquire permits on the given semaphore within a given allowed timeout and deal with errors.\n     *\n     * @param permits the mumber of permits to acquire.\n     * @param maxConcurrentRequests the maximum number of permits the semaphore was initialized\n     *     with.\n     * @param maxConcurrentRequestsTimeout the timeout to acquire the permits.\n     * @param semaphore the semaphore to acquire permits to.\n     * @throws InterruptedException if the current thread was interrupted.\n     * @throws TimeoutException if the waiting time elapsed before all permits were acquired.\n     */\n    public static void tryAcquire(\n            int permits,\n            int maxConcurrentRequests,\n            Duration maxConcurrentRequestsTimeout,\n            Semaphore semaphore)\n            throws InterruptedException, TimeoutException {\n        if (!semaphore.tryAcquire(\n                permits, maxConcurrentRequestsTimeout.toMillis(), TimeUnit.MILLISECONDS)) {\n            throw new TimeoutException(\n                    String.format(\n                            \"Failed to acquire %d out of %d permits to send value in %s.\",\n                            permits, maxConcurrentRequests, maxConcurrentRequestsTimeout));\n        }\n    }\n}\n","sourceCodeStart":34,"sourceCodeEnd":59,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/io/SinkUtils.java#L34-L59","documentation":"Thrown by SinkUtils.tryAcquire() as a java.util.concurrent.TimeoutException when semaphore.tryAcquire(permits, timeout) returns false within the configured timeout. The semaphore caps concurrent in-flight requests; if existing permits are not released quickly enough the requested number of permits cannot be obtained before maxConcurrentRequestsTimeout elapses.","triggerScenarios":"Calling SinkUtils.tryAcquire() with a maxConcurrentRequestsTimeout that is too short for the workload, or when downstream backpressure prevents the sink from releasing permits (releasing happens on completion callbacks). Also triggered if permits is larger than the semaphore's total capacity (maxConcurrentRequests).","commonSituations":"An async sink (e.g., a custom AsyncSinkBase subclass) is configured with a very low maxConcurrentRequests or a very short maxConcurrentRequestsTimeout. Network latency or a slow downstream system (database, REST API) causes in-flight requests to hold permits longer than the timeout. Misconfiguration where maxConcurrentRequestsTimeout is set to a near-zero Duration.","solutions":["Increase maxConcurrentRequestsTimeout to accommodate the worst-case round-trip time of the downstream system.","Increase maxConcurrentRequests so more permits are available, reducing contention.","Investigate downstream latency: if requests are slow to complete, the permits will not be released in time. Optimize the sink target or reduce batch size.","Ensure that permits are always released in both success and error completion paths of async requests."],"exampleFix":"// before\nSinkUtils.tryAcquire(1, 5, Duration.ofMillis(10), semaphore);\n// after — give enough headroom for downstream latency\nSinkUtils.tryAcquire(1, 10, Duration.ofSeconds(5), semaphore);","handlingStrategy":"try-catch","validationCode":"// Validate config before using the sink\nDuration timeout = maxConcurrentRequestsTimeout;\nif (timeout == null || timeout.isZero() || timeout.isNegative()) {\n    throw new IllegalArgumentException(\"maxConcurrentRequestsTimeout must be positive\");\n}\nif (permits > maxConcurrentRequests) {\n    throw new IllegalArgumentException(\"permits cannot exceed maxConcurrentRequests\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    SinkUtils.tryAcquire(permits, maxConcurrentRequests, timeout, semaphore);\n} catch (TimeoutException e) {\n    // Permits not available in time: back off, reduce batch size, or fail the request\n    LOG.warn(\"Could not acquire {} permits within {}\", permits, timeout);\n    throw e; // or apply a fallback strategy\n} catch (InterruptedException e) {\n    Thread.currentThread().interrupt();\n    throw new RuntimeException(\"Interrupted while acquiring sink permits\", e);\n}","preventionTips":["Size maxConcurrentRequestsTimeout to accommodate worst-case downstream latency plus a safety margin.","Always release permits in both success and error completion callbacks to avoid permit leaks.","Monitor semaphore queue length as an early warning for backpressure."],"tags":["async-sink","rate-limiting","semaphore","timeout","backpressure"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}