From 538b6b7ffd5ec0f98dc902a376d65d957143dcf8 Mon Sep 17 00:00:00 2001 From: rongtong Date: Fri, 24 Sep 2021 17:55:02 +0800 Subject: [PATCH] [ISSUE #1869] Delay message can't be consumed when delay offset in delayOffset.json is wrong (#3358) * fix(store):delay message can't be consumed when delay offset is wrong * fix(store):delay message can't be consumed when delay offset is wrong * fix(common):pass the code style * fix(store): fix bug of get cq max offset * fix(store): polish the code * fix(store): fix bug to pass the UT test * chore(store): polish log format * chore(store): polish code style --- .../rocketmq/common/stats/StatsItem.java | 1 - .../rocketmq/store/DefaultMessageStore.java | 9 +- .../schedule/ScheduleMessageService.java | 64 +++++++++++--- .../store/ScheduleMessageServiceTest.java | 84 +++++++++++++++++++ 4 files changed, 141 insertions(+), 17 deletions(-) create mode 100644 store/src/test/java/org/apache/rocketmq/store/ScheduleMessageServiceTest.java diff --git a/common/src/main/java/org/apache/rocketmq/common/stats/StatsItem.java b/common/src/main/java/org/apache/rocketmq/common/stats/StatsItem.java index 6007cb0a..d016662a 100644 --- a/common/src/main/java/org/apache/rocketmq/common/stats/StatsItem.java +++ b/common/src/main/java/org/apache/rocketmq/common/stats/StatsItem.java @@ -20,7 +20,6 @@ package org.apache.rocketmq.common.stats; import java.util.LinkedList; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.LongAdder; import org.apache.rocketmq.common.UtilAll; diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java index 5bf68ac5..81344c0e 100644 --- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java @@ -186,10 +186,6 @@ public class DefaultMessageStore implements MessageStore { boolean lastExitOK = !this.isTempFileExist(); log.info("last shutdown {}", lastExitOK ? "normally" : "abnormally"); - if (null != scheduleMessageService) { - result = result && this.scheduleMessageService.load(); - } - // load Commit Log result = result && this.commitLog.load(); @@ -205,7 +201,12 @@ public class DefaultMessageStore implements MessageStore { this.recover(lastExitOK); log.info("load over, and the max phy offset = {}", this.getMaxPhyOffset()); + + if (null != scheduleMessageService) { + result = this.scheduleMessageService.load(); + } } + } catch (Exception e) { log.error("load exception", e); result = false; diff --git a/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java b/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java index 1164ab8f..e0e7b956 100644 --- a/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java +++ b/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java @@ -78,8 +78,7 @@ public class ScheduleMessageService extends ConfigManager { } /** - * @param writeMessageStore - * the writeMessageStore to set + * @param writeMessageStore the writeMessageStore to set */ public void setWriteMessageStore(MessageStore writeMessageStore) { this.writeMessageStore = writeMessageStore; @@ -133,7 +132,9 @@ public class ScheduleMessageService extends ConfigManager { @Override public void run() { try { - if (started.get()) ScheduleMessageService.this.persist(); + if (started.get()) { + ScheduleMessageService.this.persist(); + } } catch (Throwable e) { log.error("scheduleAtFixedRate flush exception", e); } @@ -165,9 +166,46 @@ public class ScheduleMessageService extends ConfigManager { public boolean load() { boolean result = super.load(); result = result && this.parseDelayLevel(); + result = result && this.correctDelayOffset(); return result; } + public boolean correctDelayOffset() { + try { + for (int delayLevel : delayLevelTable.keySet()) { + ConsumeQueue cq = + ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, + delayLevel2QueueId(delayLevel)); + Long currentDelayOffset = offsetTable.get(delayLevel); + if (currentDelayOffset == null || cq == null) { + continue; + } + long correctDelayOffset = currentDelayOffset; + long cqMinOffset = cq.getMinOffsetInQueue(); + long cqMaxOffset = cq.getMaxOffsetInQueue(); + if (currentDelayOffset < cqMinOffset) { + correctDelayOffset = cqMinOffset; + log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, cqMaxOffset={}, queueId={}", + currentDelayOffset, cqMinOffset, cqMaxOffset, cq.getQueueId()); + } + + if (currentDelayOffset > cqMaxOffset) { + correctDelayOffset = cqMaxOffset; + log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, cqMaxOffset={}, queueId={}", + currentDelayOffset, cqMinOffset, cqMaxOffset, cq.getQueueId()); + } + if (correctDelayOffset != currentDelayOffset) { + log.error("correct delay offset [ delayLevel {} ] from {} to {}", delayLevel, currentDelayOffset, correctDelayOffset); + offsetTable.put(delayLevel, correctDelayOffset); + } + } + } catch (Exception e) { + log.error("correctDelayOffset exception", e); + return false; + } + return true; + } + @Override public String configFilePath() { return StorePathConfigHelper.getDelayOffsetStorePath(this.defaultMessageStore.getMessageStoreConfig() @@ -309,7 +347,7 @@ public class ScheduleMessageService extends ConfigManager { MessageExtBrokerInner msgInner = this.messageTimeup(msgExt); if (TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC.equals(msgInner.getTopic())) { log.error("[BUG] the real topic of schedule msg is {}, discard the msg. msg={}", - msgInner.getTopic(), msgInner); + msgInner.getTopic(), msgInner); continue; } PutMessageResult putMessageResult = @@ -344,14 +382,9 @@ public class ScheduleMessageService extends ConfigManager { } catch (Exception e) { /* * XXX: warn and notify me - - - */ log.error( - "ScheduleMessageService, messageTimeup execute error, drop it. msgExt=" - + msgExt + ", nextOffset=" + nextOffset + ",offsetPy=" - + offsetPy + ",sizePy=" + sizePy, e); + "ScheduleMessageService, messageTimeup execute error, drop it. msgExt={}, nextOffset={}, offsetPy={}, sizePy={}", msgExt, nextOffset, offsetPy, sizePy, e); } } } else { @@ -376,10 +409,17 @@ public class ScheduleMessageService extends ConfigManager { else { long cqMinOffset = cq.getMinOffsetInQueue(); + long cqMaxOffset = cq.getMaxOffsetInQueue(); if (offset < cqMinOffset) { failScheduleOffset = cqMinOffset; - log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset=" - + cqMinOffset + ", queueId=" + cq.getQueueId()); + log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, cqMaxOffset={}, queueId={}", + offset, cqMinOffset, cqMaxOffset, cq.getQueueId()); + } + + if (offset > cqMaxOffset) { + failScheduleOffset = cqMaxOffset; + log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, cqMaxOffset={}, queueId={}", + offset, cqMinOffset, cqMaxOffset, cq.getQueueId()); } } } // end of if (cq != null) diff --git a/store/src/test/java/org/apache/rocketmq/store/ScheduleMessageServiceTest.java b/store/src/test/java/org/apache/rocketmq/store/ScheduleMessageServiceTest.java new file mode 100644 index 00000000..8502521f --- /dev/null +++ b/store/src/test/java/org/apache/rocketmq/store/ScheduleMessageServiceTest.java @@ -0,0 +1,84 @@ +/* + * 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.store; + +import java.lang.reflect.Field; +import java.util.Map; +import java.util.Random; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.store.config.FlushDiskType; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.schedule.ScheduleMessageService; +import org.apache.rocketmq.store.stats.BrokerStatsManager; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class ScheduleMessageServiceTest { + + private Random random = new Random(); + + @Test + public void testCorrectDelayOffset_whenInit() throws Exception { + + ConcurrentMap offsetTable = null; + + ScheduleMessageService scheduleMessageService = new ScheduleMessageService((DefaultMessageStore) buildMessageStore()); + scheduleMessageService.parseDelayLevel(); + + ConcurrentMap offsetTable1 = new ConcurrentHashMap<>(); + for (int i = 1; i <= 18; i++) { + offsetTable1.put(i, random.nextLong()); + } + + Field field = scheduleMessageService.getClass().getDeclaredField("offsetTable"); + field.setAccessible(true); + field.set(scheduleMessageService, offsetTable1); + + String jsonStr = scheduleMessageService.encode(); + scheduleMessageService.decode(jsonStr); + + offsetTable = (ConcurrentMap) field.get(scheduleMessageService); + + for (Map.Entry entry : offsetTable.entrySet()) { + assertEquals(entry.getValue(), offsetTable1.get(entry.getKey())); + } + + scheduleMessageService.correctDelayOffset(); + + offsetTable = (ConcurrentMap) field.get(scheduleMessageService); + + for (long offset : offsetTable.values()) { + assertEquals(offset, 0); + } + + } + + private MessageStore buildMessageStore() throws Exception { + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + messageStoreConfig.setMappedFileSizeCommitLog(1024 * 1024 * 10); + messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 1024 * 10); + messageStoreConfig.setMaxHashSlotNum(10000); + messageStoreConfig.setMaxIndexNum(100 * 100); + messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH); + messageStoreConfig.setFlushIntervalConsumeQueue(1); + return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest"), null, new BrokerConfig()); + } +} -- GitLab