diff --git a/broker/src/main/java/org/apache/rocketmq/broker/util/HookUtils.java b/broker/src/main/java/org/apache/rocketmq/broker/util/HookUtils.java index 94be46ea405..06eb90b6eaf 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/util/HookUtils.java +++ b/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); } diff --git a/broker/src/test/java/org/apache/rocketmq/broker/util/HookUtilsTimerOverflowTest.java b/broker/src/test/java/org/apache/rocketmq/broker/util/HookUtilsTimerOverflowTest.java new file mode 100644 index 00000000000..9c107b9b7aa --- /dev/null +++ b/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()); + } +}