[ISSUE #9923] Transactional messages should not send custom delayed messages (#9924)

* [ISSUE #9923] Transactional messages should not send custom delayed messages

* Transactional messages do not support delayed delivery

Author: yx9o

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

@@ -1434,11 +1434,7 @@ public TransactionSendResult sendMessageInTransaction(final Message msg,
             throw new MQClientException("tranExecutor is null", null);
         }
 
-        // ignore DelayTimeLevel parameter
-        if (msg.getDelayTimeLevel() != 0) {
-            MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
-        }
-
+        ensureNotDelayedForTransactional(msg);
         Validators.checkMessage(msg, this.defaultMQProducer);
 
         SendResult sendResult = null;
@@ -1495,7 +1491,7 @@ public TransactionSendResult sendMessageInTransaction(final Message msg,
         try {
             this.endTransaction(msg, sendResult, localTransactionState, localException);
         } catch (Exception e) {
-            log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
+            log.warn("local transaction execute {}, but end broker transaction failed", localTransactionState, e);
         }
 
         TransactionSendResult transactionSendResult = new TransactionSendResult();
@@ -1508,6 +1504,15 @@ public TransactionSendResult sendMessageInTransaction(final Message msg,
         return transactionSendResult;
     }
 
+    private void ensureNotDelayedForTransactional(final Message msg) throws MQClientException {
+        if (msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null
+                || msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_MS) != null
+                || msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_SEC) != null
+                || msg.getProperty(MessageConst.PROPERTY_TIMER_DELIVER_MS) != null) {
+            throw new MQClientException("Transactional messages do not support delayed delivery", null);
+        }
+    }
+
     /**
      * DEFAULT SYNC -------------------------------------------------------
      */

client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithTraceTest.java

@@ -168,6 +168,18 @@ public Object answer(InvocationOnMock mock) throws Throwable {
         assertThat(ctx.getMessage().getTopic()).isEqualTo(topic);
     }
 
+    @Test(expected = MQClientException.class)
+    public void testSendMessageInTransaction_NoListener_ThrowsException() throws MQClientException {
+        producer.setTransactionListener(null);
+        producer.sendMessageInTransaction(message, null);
+    }
+
+    @Test(expected = MQClientException.class)
+    public void testSendMessageInTransaction_DelayMsg_ThrowsException() throws MQClientException {
+        message.setDelayTimeLevel(3);
+        producer.sendMessageInTransaction(message, null);
+    }
+
     @After
     public void terminate() {
         producer.shutdown();