[ISSUE #10260] Reject delayed transactional messages in gRPC send path (#10261)

Author: yx9o

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java

@@ -249,6 +249,9 @@ protected Map<String, String> buildMessageProperty(ProxyContext context, apache.
         // set transaction property
         MessageType messageType = message.getSystemProperties().getMessageType();
         if (messageType.equals(MessageType.TRANSACTION)) {
+            if (message.getSystemProperties().hasDeliveryTimestamp()) {
+                throw new GrpcProxyException(Code.BAD_REQUEST, "transaction message cannot set delivery timestamp");
+            }
             MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
 
             if (message.getSystemProperties().hasOrphanedTransactionRecoveryDuration()) {

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java

@@ -925,6 +925,36 @@ public void testParameterValidate() {
             }
         });
 
+        // transaction message cannot be delay message
+        assertThrows(GrpcProxyException.class, () -> {
+            try {
+                this.sendMessageActivity.sendMessage(
+                    createContext(),
+                    SendMessageRequest.newBuilder()
+                        .addMessages(Message.newBuilder()
+                            .setTopic(Resource.newBuilder()
+                                .setName(TOPIC)
+                                .build())
+                            .setSystemProperties(SystemProperties.newBuilder()
+                                .setMessageId("id")
+                                .setDeliveryTimestamp(Timestamps.fromMillis(System.currentTimeMillis() + Duration.ofSeconds(5).toMillis()))
+                                .setQueueId(0)
+                                .setMessageType(MessageType.TRANSACTION)
+                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
+                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
+                                .build())
+                            .setBody(ByteString.copyFrom(new byte[3]))
+                            .build())
+                        .build()
+                ).get();
+            } catch (ExecutionException t) {
+                GrpcProxyException e = (GrpcProxyException) t.getCause();
+                assertEquals(Code.BAD_REQUEST, e.getCode());
+                assertEquals("transaction message cannot set delivery timestamp", e.getMessage());
+                throw e;
+            }
+        });
+
         // transactionRecoverySecond
         assertThrows(GrpcProxyException.class, () -> {
             try {