fix(broker): reject timer delays that overflow the absolute delivery time

broker/src/main/java/org/apache/rocketmq/broker/util/HookUtils.java

@@ -206,19 +206,23 @@ private static PutMessageResult transformTimerMessage(BrokerController brokerCon
         //do transform
         int delayLevel = msg.getDelayTimeLevel();
         long deliverMs;
+        long now = System.currentTimeMillis();
         try {
             if (msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_SEC) != null) {
-                deliverMs = System.currentTimeMillis() + Long.parseLong(msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_SEC)) * 1000;
+                long delaySec = Long.parseLong(msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_SEC));
+                deliverMs = Math.multiplyExact(delaySec, 1000L);
+                deliverMs = Math.addExact(now, deliverMs);
             } else if (msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_MS) != null) {
-                deliverMs = System.currentTimeMillis() + Long.parseLong(msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_MS));
+                long delayMs = Long.parseLong(msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_MS));
+                deliverMs = Math.addExact(now, delayMs);
             } else {
                 deliverMs = Long.parseLong(msg.getProperty(MessageConst.PROPERTY_TIMER_DELIVER_MS));
             }
         } catch (Exception e) {
             return new PutMessageResult(PutMessageStatus.WHEEL_TIMER_MSG_ILLEGAL, null);
         }
-        if (deliverMs > System.currentTimeMillis()) {
-            if (delayLevel <= 0 && deliverMs - System.currentTimeMillis() > brokerController.getMessageStoreConfig().getTimerMaxDelaySec() * 1000L) {
+        if (deliverMs > now) {
+            if (delayLevel <= 0 && deliverMs - now > brokerController.getMessageStoreConfig().getTimerMaxDelaySec() * 1000L) {
                 return new PutMessageResult(PutMessageStatus.WHEEL_TIMER_MSG_ILLEGAL, null);
             }
 

broker/src/test/java/org/apache/rocketmq/broker/util/HookUtilsTimerOverflowTest.java

@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.util;
+
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageExtBrokerInner;
+import org.apache.rocketmq.store.PutMessageResult;
+import org.apache.rocketmq.store.PutMessageStatus;
+import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.apache.rocketmq.store.timer.TimerMessageStore;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class HookUtilsTimerOverflowTest {
+
+    private BrokerController newTimerEnabledController() {
+        BrokerController brokerController = Mockito.mock(BrokerController.class);
+        MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+        messageStoreConfig.setTimerWheelEnable(true);
+        Mockito.when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig);
+        return brokerController;
+    }
+
+    private MessageExtBrokerInner newTimerMessage(String key, String value) {
+        MessageExtBrokerInner msg = new MessageExtBrokerInner();
+        msg.setTopic("OverflowTestTopic");
+        MessageAccessor.putProperty(msg, key, value);
+        return msg;
+    }
+
+    // Regression for https://github.com/apache/rocketmq/issues/10872: an
+    // overflowing relative delay used to wrap around into a past timestamp,
+    // bypass the future-delay validation, and take the immediate-message path.
+    @Test
+    public void testTimerDelaySecOverflowRejected() {
+        PutMessageResult result = HookUtils.handleScheduleMessage(newTimerEnabledController(),
+            newTimerMessage(MessageConst.PROPERTY_TIMER_DELAY_SEC, String.valueOf(Long.MAX_VALUE)));
+        Assert.assertNotNull(result);
+        Assert.assertEquals(PutMessageStatus.WHEEL_TIMER_MSG_ILLEGAL, result.getPutMessageStatus());
+    }
+
+    @Test
+    public void testTimerDelayMsOverflowRejected() {
+        PutMessageResult result = HookUtils.handleScheduleMessage(newTimerEnabledController(),
+            newTimerMessage(MessageConst.PROPERTY_TIMER_DELAY_MS, String.valueOf(Long.MAX_VALUE)));
+        Assert.assertNotNull(result);
+        Assert.assertEquals(PutMessageStatus.WHEEL_TIMER_MSG_ILLEGAL, result.getPutMessageStatus());
+    }
+
+    // Long.MAX_VALUE / 1000 keeps multiplyExact within range but overflows the
+    // subsequent addExact against the current time, exercising its throw path.
+    @Test
+    public void testTimerDelaySecAddExactOverflowRejected() {
+        PutMessageResult result = HookUtils.handleScheduleMessage(newTimerEnabledController(),
+            newTimerMessage(MessageConst.PROPERTY_TIMER_DELAY_SEC, String.valueOf(Long.MAX_VALUE / 1000)));
+        Assert.assertNotNull(result);
+        Assert.assertEquals(PutMessageStatus.WHEEL_TIMER_MSG_ILLEGAL, result.getPutMessageStatus());
+    }
+
+    // Exercises the successful timer transformation for a millisecond delay:
+    // the message is rewritten to the wheel-timer topic with the normalized
+    // delivery time stored in PROPERTY_TIMER_OUT_MS.
+    @Test
+    public void testTimerDelayMsSuccessPath() {
+        BrokerController brokerController = newTimerEnabledController();
+        TimerMessageStore timerMessageStore = Mockito.mock(TimerMessageStore.class);
+        Mockito.when(brokerController.getTimerMessageStore()).thenReturn(timerMessageStore);
+        Mockito.when(timerMessageStore.isReject(Mockito.anyLong())).thenReturn(false);
+
+        MessageExtBrokerInner msg = newTimerMessage(MessageConst.PROPERTY_TIMER_DELAY_MS, "1000");
+        PutMessageResult result = HookUtils.handleScheduleMessage(brokerController, msg);
+        Assert.assertNull(result);
+        Assert.assertNotNull(msg.getProperty(MessageConst.PROPERTY_TIMER_OUT_MS));
+        Assert.assertEquals(TimerMessageStore.TIMER_TOPIC, msg.getTopic());
+    }
+
+    // Covers the future-delivery validation branch with a delay that does not
+    // overflow but exceeds the configured maximum.
+    @Test
+    public void testTimerDelaySecExceedsMaxRejected() {
+        BrokerController brokerController = newTimerEnabledController();
+        brokerController.getMessageStoreConfig().setTimerMaxDelaySec(1);
+        PutMessageResult result = HookUtils.handleScheduleMessage(brokerController,
+            newTimerMessage(MessageConst.PROPERTY_TIMER_DELAY_SEC, "10"));
+        Assert.assertNotNull(result);
+        Assert.assertEquals(PutMessageStatus.WHEEL_TIMER_MSG_ILLEGAL, result.getPutMessageStatus());
+    }
+}