[ISSUE #8997] Ensure there is an opportunity to send a retry message when broker no response (#9137)

Author: gaoyf

client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java

@@ -777,8 +777,16 @@ private SendResult sendDefaultImpl(
                             callTimeout = true;
                             break;
                         }
-
-                        sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
+                        long curTimeout = timeout - costTime;
+                        // Get the maximum timeout allowed per request
+                        long maxSendTimeoutPerRequest = defaultMQProducer.getSendMsgMaxTimeoutPerRequest();
+                        // Determine if retries are still possible
+                        boolean canRetryAgain = times + 1 < timesTotal;
+                        // If retries are possible, and the current timeout exceeds the max allowed timeout, set the current timeout to the max allowed
+                        if (maxSendTimeoutPerRequest > -1 && canRetryAgain && curTimeout > maxSendTimeoutPerRequest) {
+                            curTimeout = maxSendTimeoutPerRequest;
+                        }
+                        sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, curTimeout);
                         endTimestamp = System.currentTimeMillis();
                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false, true);
                         switch (communicationMode) {

client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java

@@ -115,6 +115,11 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
      */
     private int sendMsgTimeout = 3000;
 
+    /**
+     * Max timeout for sending messages per request.
+     */
+    private int sendMsgMaxTimeoutPerRequest = -1;
+
     /**
      * Compress message body threshold, namely, message body larger than 4k will be compressed on default.
      */
@@ -1259,6 +1264,14 @@ public void setSendMsgTimeout(int sendMsgTimeout) {
         this.sendMsgTimeout = sendMsgTimeout;
     }
 
+    public int getSendMsgMaxTimeoutPerRequest() {
+        return sendMsgMaxTimeoutPerRequest;
+    }
+
+    public void setSendMsgMaxTimeoutPerRequest(int sendMsgMaxTimeoutPerRequest) {
+        this.sendMsgMaxTimeoutPerRequest = sendMsgMaxTimeoutPerRequest;
+    }
+
     public int getCompressMsgBodyOverHowmuch() {
         return compressMsgBodyOverHowmuch;
     }