{"record":{"id":"79b51e655e9ab0da","repo":"apache/rocketmq","slug":"executor-rejected","errorCode":null,"errorMessage":"executor rejected ","messagePattern":"executor rejected ","errorType":"exception","errorClass":"MQClientException","httpStatus":null,"severity":"error","filePath":"client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java","lineNumber":679,"sourceCode":"                costTime = System.currentTimeMillis() - beginStartTime;\n\n                isSemaphoreAsyncSizeAcquired = timeout - costTime > 0\n                    && semaphoreAsyncSendSize.tryAcquire(msgLen, timeout - costTime, TimeUnit.MILLISECONDS);\n                sendCallback.isSemaphoreAsyncSizeAcquired = isSemaphoreAsyncSizeAcquired;\n                defaultMQProducer.releaseBackPressureForAsyncSendSizeLock();\n                if (!isSemaphoreAsyncSizeAcquired) {\n                    sendCallback.onException(\n                        new RemotingTooMuchRequestException(\"send message tryAcquire semaphoreAsyncSize timeout\"));\n                    return;\n                }\n            }\n\n            executor.submit(runnable);\n        } catch (RejectedExecutionException e) {\n            if (isEnableBackpressureForAsyncMode) {\n                runnable.run();\n            } else {\n                throw new MQClientException(\"executor rejected \", e);\n            }\n        }\n    }\n\n    public MessageQueue invokeMessageQueueSelector(Message msg, MessageQueueSelector selector, Object arg,\n                                                   final long timeout) throws MQClientException, RemotingTooMuchRequestException {\n        long beginStartTime = System.currentTimeMillis();\n        this.makeSureStateOK();\n        Validators.checkMessage(msg, this.defaultMQProducer);\n\n        TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());\n        if (topicPublishInfo != null && topicPublishInfo.ok()) {\n            MessageQueue mq = null;\n            try {\n                List<MessageQueue> messageQueueList =\n                        mQClientFactory.getMQAdminImpl().parsePublishMessageQueues(topicPublishInfo.getMessageQueueList());\n                Message userMessage = MessageAccessor.cloneMessage(msg);\n                String userTopic = NamespaceUtil.withoutNamespace(userMessage.getTopic(), mQClientFactory.getClientConfig().getNamespace());","sourceCodeStart":661,"sourceCodeEnd":697,"githubUrl":"https://github.com/apache/rocketmq/blob/293f5885719fc4aa3619446a1900f58ccfcfdd29/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java#L661-L697","documentation":"MQClientException thrown from the async-send path when executor.submit(runnable) raises RejectedExecutionException and backpressure is disabled. The default async executor has a bounded queue; once the in-flight async sends exceed it, the JDK executor rejects the task and RocketMQ rethrows it as this error. When enableBackpressureForAsyncMode is true the rejection is instead absorbed by running the task on the caller thread (natural backpressure).","triggerScenarios":"triggerScenarios","commonSituations":"See trigger scenarios.","solutions":["Enable backpressure so rejection degrades to caller-thread execution: producer.setEnableBackpressureForAsyncMode(true) (and tune backpressureForAsyncFlushTimeout if needed)","Increase the async executor queue capacity via producer.setAsyncExecutorSizeConfig(...) / defaultAsyncExecutorQueueSize to match peak throughput","Rate-limit or batch sends at the application layer (semaphore, Resilience4j bulkhead) so in-flight async sends stay under the queue bound","Ensure SendCallback implementations return quickly; offload heavy work to a separate executor"],"exampleFix":"// before\nproducer.setEnableBackpressureForAsyncMode(false);\nproducer.send(msg, callback); // burst -> executor rejected\n// after\nproducer.setEnableBackpressureForAsyncMode(true);\nproducer.send(msg, callback); // overflow runs on caller thread instead of throwing","handlingStrategy":"fallback","validationCode":" producer.setEnableBackpressureForAsyncMode(true); // rejection degrades to caller-thread run\n// optional: size the executor for peak throughput\nproducer.getDefaultMQProducerImpl()... // or set asyncExecutorSizeConfig queue size >= peak in-flight sends","typeGuard":null,"tryCatchPattern":"producer.send(msg, new SendCallback() {\n    public void onSuccess(SendResult r) { ... }\n    public void onException(Throwable e) {\n        if (e instanceof MQClientException && String.valueOf(e.getMessage()).contains(\"executor rejected\")) {\n            localBuffer.offer(msg); // fallback: spool and retry later\n        }\n    }\n});","preventionTips":["Enable enableBackpressureForAsyncMode in high-throughput async producers","Bound in-flight async sends with an application-side semaphore (e.g. Semaphore(N)) before calling send","Size the async executor queue from load tests (peak RPS * worst-case latency)","Keep SendCallback bodies fast; offload heavy processing to another pool"],"tags":["rocketmq","producer","async","backpressure","thread-pool"],"backgroundTag":null,"analyzedSha":"293f5885719fc4aa3619446a1900f58ccfcfdd29","analyzedAt":"2026-08-14T11:50:13.822Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}