majialoong commented on code in PR #9787: URL: https://github.com/apache/rocketmq/pull/9787#discussion_r2478396430
########## store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java: ########## @@ -0,0 +1,619 @@ +/* + * 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.timer.rocksdb; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import com.conversantmedia.util.concurrent.DisruptorBlockingQueue; +import com.google.common.util.concurrent.RateLimiter; +import io.opentelemetry.api.common.Attributes; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.ServiceThread; +import org.apache.rocketmq.common.TopicFilterType; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.message.MessageAccessor; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageExtBrokerInner; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; +import org.apache.rocketmq.store.DefaultMessageStore; +import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.PutMessageResult; +import org.apache.rocketmq.store.StoreUtil; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsConstant; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsManager; +import org.apache.rocketmq.store.metrics.StoreMetricsManager; +import org.apache.rocketmq.store.queue.ConsumeQueueInterface; +import org.apache.rocketmq.store.queue.CqUnit; +import org.apache.rocketmq.store.queue.ReferredIterator; +import org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage; +import org.apache.rocketmq.store.stats.BrokerStatsManager; +import org.apache.rocketmq.store.timer.TimerMetrics; +import org.apache.rocketmq.store.util.PerfCounter; +import static org.apache.rocketmq.common.message.MessageConst.PROPERTY_TIMER_ROLL_LABEL; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMER_COLUMN_FAMILY; +import static org.apache.rocketmq.store.timer.TimerMessageStore.TIMER_TOPIC; + +public class TimerMessageRocksDBStore { + private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); + private static final Logger logError = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME); + private static final String SCAN_SYS_TOPIC = "scanSysTopic"; + private static final String SCAN_SYS_TOPIC_MISS = "scanSysTopicMiss"; + private static final String OUT_BIZ_MESSAGE = "outBizMsg"; + private static final String ROLL_LABEL = "R"; + private static final int PUT_OK = 0, PUT_NEED_RETRY = 1, PUT_NO_RETRY = 2; + private static final int MAX_GET_MSG_TIMES = 3, MAX_PUT_MSG_TIMES = 3; + private static final int TIME_UP_CAPACITY = 100, ROLL_CAPACITY = 50; + private static final int INITIAL = 0, RUNNING = 1, SHUTDOWN = 2; + private volatile int state = INITIAL; + private static long expirationThresholdMillis = 999L; + private final AtomicLong readOffset = new AtomicLong(0);//timerSysTopic read offset + private final MessageStore messageStore; + private final TimerMetrics timerMetrics; + private final MessageStoreConfig storeConfig; + private final BrokerStatsManager brokerStatsManager; + private final MessageRocksDBStorage messageRocksDBStorage; + private Timeline timeline; + private TimerSysTopicScanService timerSysTopicScanService; + private TimerMessageReputService expiredMessageReputService; + private TimerMessageReputService rollMessageReputService; + protected long precisionMs; + private BlockingQueue<List<TimerRocksDBRecord>> expiredMessageQueue; + private BlockingQueue<List<TimerRocksDBRecord>> rollMessageQueue; + protected final PerfCounter.Ticks perfCounterTicks = new PerfCounter.Ticks(log); + private Function<MessageExtBrokerInner, PutMessageResult> escapeBridgeHook; + private ThreadLocal<ByteBuffer> bufferLocal = null; + + public TimerMessageRocksDBStore(final MessageStore messageStore, final TimerMetrics timerMetrics, + final BrokerStatsManager brokerStatsManager) { + this.messageStore = messageStore; + this.storeConfig = messageStore.getMessageStoreConfig(); + this.precisionMs = storeConfig.getTimerRocksDBPrecisionMs() < 100L ? 1000L : storeConfig.getTimerRocksDBPrecisionMs(); + expirationThresholdMillis = precisionMs - 1L; + this.messageRocksDBStorage = messageStore.getMessageRocksDBStorage(); + this.timerMetrics = timerMetrics; + this.brokerStatsManager = brokerStatsManager; + bufferLocal = ThreadLocal.withInitial(() -> ByteBuffer.allocate(storeConfig.getMaxMessageSize())); + } + + public synchronized boolean load() { + initService(); + boolean result = this.timerMetrics.load(); + log.info("TimerMessageRocksDBStore load result: {}", result); + return result; + } + + public synchronized void start() { + if (this.state == RUNNING) { + return; + } + long commitOffsetFile = null != this.messageStore.getTimerMessageStore() ? this.messageStore.getTimerMessageStore().getCommitQueueOffset() : 0L; + long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); + long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB); + this.readOffset.set(maxCommitOffset); + this.expiredMessageReputService.start(); + this.rollMessageReputService.start(); + this.timeline.start(); + this.timerSysTopicScanService.start(); + this.state = RUNNING; + log.info("TimerMessageRocksDBStore start success, start commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, this.readOffset.get()); + } + + public synchronized boolean restart() { + try { + this.storeConfig.setTimerStopEnqueue(true); + if (this.state == RUNNING && !this.storeConfig.isTimerRocksDBStopScan()) { + log.info("restart TimerMessageRocksDBStore has been running"); + return true; + } + if (this.state == RUNNING && this.storeConfig.isTimerRocksDBStopScan()) { + long commitOffsetFile = null != this.messageStore.getTimerMessageStore() ? this.messageStore.getTimerMessageStore().getCommitQueueOffset() : 0L; + long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); + long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB); + this.readOffset.set(maxCommitOffset); + log.info("restart TimerMessageRocksDBStore has benn recover running, commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, readOffset.get()); Review Comment: maybe is `has been recovered to running` ########## store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java: ########## @@ -0,0 +1,619 @@ +/* + * 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.timer.rocksdb; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import com.conversantmedia.util.concurrent.DisruptorBlockingQueue; +import com.google.common.util.concurrent.RateLimiter; +import io.opentelemetry.api.common.Attributes; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.ServiceThread; +import org.apache.rocketmq.common.TopicFilterType; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.message.MessageAccessor; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageExtBrokerInner; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; +import org.apache.rocketmq.store.DefaultMessageStore; +import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.PutMessageResult; +import org.apache.rocketmq.store.StoreUtil; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsConstant; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsManager; +import org.apache.rocketmq.store.metrics.StoreMetricsManager; +import org.apache.rocketmq.store.queue.ConsumeQueueInterface; +import org.apache.rocketmq.store.queue.CqUnit; +import org.apache.rocketmq.store.queue.ReferredIterator; +import org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage; +import org.apache.rocketmq.store.stats.BrokerStatsManager; +import org.apache.rocketmq.store.timer.TimerMetrics; +import org.apache.rocketmq.store.util.PerfCounter; +import static org.apache.rocketmq.common.message.MessageConst.PROPERTY_TIMER_ROLL_LABEL; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMER_COLUMN_FAMILY; +import static org.apache.rocketmq.store.timer.TimerMessageStore.TIMER_TOPIC; + +public class TimerMessageRocksDBStore { + private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); + private static final Logger logError = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME); + private static final String SCAN_SYS_TOPIC = "scanSysTopic"; + private static final String SCAN_SYS_TOPIC_MISS = "scanSysTopicMiss"; + private static final String OUT_BIZ_MESSAGE = "outBizMsg"; + private static final String ROLL_LABEL = "R"; + private static final int PUT_OK = 0, PUT_NEED_RETRY = 1, PUT_NO_RETRY = 2; + private static final int MAX_GET_MSG_TIMES = 3, MAX_PUT_MSG_TIMES = 3; + private static final int TIME_UP_CAPACITY = 100, ROLL_CAPACITY = 50; + private static final int INITIAL = 0, RUNNING = 1, SHUTDOWN = 2; + private volatile int state = INITIAL; + private static long expirationThresholdMillis = 999L; + private final AtomicLong readOffset = new AtomicLong(0);//timerSysTopic read offset + private final MessageStore messageStore; + private final TimerMetrics timerMetrics; + private final MessageStoreConfig storeConfig; + private final BrokerStatsManager brokerStatsManager; + private final MessageRocksDBStorage messageRocksDBStorage; + private Timeline timeline; + private TimerSysTopicScanService timerSysTopicScanService; + private TimerMessageReputService expiredMessageReputService; + private TimerMessageReputService rollMessageReputService; + protected long precisionMs; + private BlockingQueue<List<TimerRocksDBRecord>> expiredMessageQueue; + private BlockingQueue<List<TimerRocksDBRecord>> rollMessageQueue; + protected final PerfCounter.Ticks perfCounterTicks = new PerfCounter.Ticks(log); + private Function<MessageExtBrokerInner, PutMessageResult> escapeBridgeHook; + private ThreadLocal<ByteBuffer> bufferLocal = null; + + public TimerMessageRocksDBStore(final MessageStore messageStore, final TimerMetrics timerMetrics, + final BrokerStatsManager brokerStatsManager) { + this.messageStore = messageStore; + this.storeConfig = messageStore.getMessageStoreConfig(); + this.precisionMs = storeConfig.getTimerRocksDBPrecisionMs() < 100L ? 1000L : storeConfig.getTimerRocksDBPrecisionMs(); + expirationThresholdMillis = precisionMs - 1L; + this.messageRocksDBStorage = messageStore.getMessageRocksDBStorage(); + this.timerMetrics = timerMetrics; + this.brokerStatsManager = brokerStatsManager; + bufferLocal = ThreadLocal.withInitial(() -> ByteBuffer.allocate(storeConfig.getMaxMessageSize())); + } + + public synchronized boolean load() { + initService(); + boolean result = this.timerMetrics.load(); + log.info("TimerMessageRocksDBStore load result: {}", result); + return result; + } + + public synchronized void start() { + if (this.state == RUNNING) { + return; + } + long commitOffsetFile = null != this.messageStore.getTimerMessageStore() ? this.messageStore.getTimerMessageStore().getCommitQueueOffset() : 0L; + long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); + long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB); + this.readOffset.set(maxCommitOffset); + this.expiredMessageReputService.start(); + this.rollMessageReputService.start(); + this.timeline.start(); + this.timerSysTopicScanService.start(); + this.state = RUNNING; + log.info("TimerMessageRocksDBStore start success, start commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, this.readOffset.get()); + } + + public synchronized boolean restart() { + try { + this.storeConfig.setTimerStopEnqueue(true); + if (this.state == RUNNING && !this.storeConfig.isTimerRocksDBStopScan()) { + log.info("restart TimerMessageRocksDBStore has been running"); + return true; + } + if (this.state == RUNNING && this.storeConfig.isTimerRocksDBStopScan()) { + long commitOffsetFile = null != this.messageStore.getTimerMessageStore() ? this.messageStore.getTimerMessageStore().getCommitQueueOffset() : 0L; + long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); + long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB); + this.readOffset.set(maxCommitOffset); + log.info("restart TimerMessageRocksDBStore has benn recover running, commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, readOffset.get()); + } else { + this.load(); + this.start(); + } + this.storeConfig.setTimerRocksDBEnable(true); + this.storeConfig.setTimerRocksDBStopScan(false); + return true; + } catch (Exception e) { + logError.error("TimerMessageRocksDBStore restart error: {}", e.getMessage()); + return false; + } + } + + public void shutdown() { + if (this.state != RUNNING || this.state == SHUTDOWN) { + return; + } + if (null != this.timerSysTopicScanService) { + this.timerSysTopicScanService.shutdown(); + } + if (null != this.timeline) { + this.timeline.shutDown(); + } + if (null != this.expiredMessageReputService) { + this.expiredMessageReputService.shutdown(); + } + if (null != this.rollMessageReputService) { + this.rollMessageReputService.shutdown(); + } + this.state = SHUTDOWN; + log.info("TimerMessageRocksDBStore shutdown success"); + } + + public void putRealTopicMessage(MessageExt msg) { + if (null == msg) { + logError.error("putRealTopicMessage msg is null"); + return; + } + MessageExtBrokerInner messageExtBrokerInner = convertMessage(msg); + if (null == messageExtBrokerInner) { + logError.error("putRealTopicMessage error, messageExtBrokerInner is null"); + return; + } + doPut(messageExtBrokerInner); + } + + public MessageStore getMessageStore() { + return messageStore; + } + + public TimerMetrics getTimerMetrics() { + return timerMetrics; + } + + public BrokerStatsManager getBrokerStatsManager() { + return brokerStatsManager; + } + + public AtomicLong getReadOffset() { + return readOffset; + } + + public BlockingQueue<List<TimerRocksDBRecord>> getExpiredMessageQueue() { + return expiredMessageQueue; + } + + public BlockingQueue<List<TimerRocksDBRecord>> getRollMessageQueue() { + return rollMessageQueue; + } + + public long getCommitOffsetInRocksDB() { + if (null == messageRocksDBStorage || !storeConfig.isTransRocksDBEnable()) { Review Comment: Should this be `isTimerRocksDBEnable`? ########## store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java: ########## @@ -0,0 +1,619 @@ +/* + * 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.timer.rocksdb; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import com.conversantmedia.util.concurrent.DisruptorBlockingQueue; +import com.google.common.util.concurrent.RateLimiter; +import io.opentelemetry.api.common.Attributes; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.ServiceThread; +import org.apache.rocketmq.common.TopicFilterType; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.message.MessageAccessor; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageExtBrokerInner; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; +import org.apache.rocketmq.store.DefaultMessageStore; +import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.PutMessageResult; +import org.apache.rocketmq.store.StoreUtil; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsConstant; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsManager; +import org.apache.rocketmq.store.metrics.StoreMetricsManager; +import org.apache.rocketmq.store.queue.ConsumeQueueInterface; +import org.apache.rocketmq.store.queue.CqUnit; +import org.apache.rocketmq.store.queue.ReferredIterator; +import org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage; +import org.apache.rocketmq.store.stats.BrokerStatsManager; +import org.apache.rocketmq.store.timer.TimerMetrics; +import org.apache.rocketmq.store.util.PerfCounter; +import static org.apache.rocketmq.common.message.MessageConst.PROPERTY_TIMER_ROLL_LABEL; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMER_COLUMN_FAMILY; +import static org.apache.rocketmq.store.timer.TimerMessageStore.TIMER_TOPIC; + +public class TimerMessageRocksDBStore { + private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); + private static final Logger logError = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME); + private static final String SCAN_SYS_TOPIC = "scanSysTopic"; + private static final String SCAN_SYS_TOPIC_MISS = "scanSysTopicMiss"; + private static final String OUT_BIZ_MESSAGE = "outBizMsg"; + private static final String ROLL_LABEL = "R"; + private static final int PUT_OK = 0, PUT_NEED_RETRY = 1, PUT_NO_RETRY = 2; + private static final int MAX_GET_MSG_TIMES = 3, MAX_PUT_MSG_TIMES = 3; + private static final int TIME_UP_CAPACITY = 100, ROLL_CAPACITY = 50; + private static final int INITIAL = 0, RUNNING = 1, SHUTDOWN = 2; + private volatile int state = INITIAL; + private static long expirationThresholdMillis = 999L; + private final AtomicLong readOffset = new AtomicLong(0);//timerSysTopic read offset + private final MessageStore messageStore; + private final TimerMetrics timerMetrics; + private final MessageStoreConfig storeConfig; + private final BrokerStatsManager brokerStatsManager; + private final MessageRocksDBStorage messageRocksDBStorage; + private Timeline timeline; + private TimerSysTopicScanService timerSysTopicScanService; + private TimerMessageReputService expiredMessageReputService; + private TimerMessageReputService rollMessageReputService; + protected long precisionMs; + private BlockingQueue<List<TimerRocksDBRecord>> expiredMessageQueue; + private BlockingQueue<List<TimerRocksDBRecord>> rollMessageQueue; + protected final PerfCounter.Ticks perfCounterTicks = new PerfCounter.Ticks(log); + private Function<MessageExtBrokerInner, PutMessageResult> escapeBridgeHook; + private ThreadLocal<ByteBuffer> bufferLocal = null; + + public TimerMessageRocksDBStore(final MessageStore messageStore, final TimerMetrics timerMetrics, + final BrokerStatsManager brokerStatsManager) { + this.messageStore = messageStore; + this.storeConfig = messageStore.getMessageStoreConfig(); + this.precisionMs = storeConfig.getTimerRocksDBPrecisionMs() < 100L ? 1000L : storeConfig.getTimerRocksDBPrecisionMs(); + expirationThresholdMillis = precisionMs - 1L; + this.messageRocksDBStorage = messageStore.getMessageRocksDBStorage(); + this.timerMetrics = timerMetrics; + this.brokerStatsManager = brokerStatsManager; + bufferLocal = ThreadLocal.withInitial(() -> ByteBuffer.allocate(storeConfig.getMaxMessageSize())); + } + + public synchronized boolean load() { + initService(); + boolean result = this.timerMetrics.load(); + log.info("TimerMessageRocksDBStore load result: {}", result); + return result; + } + + public synchronized void start() { + if (this.state == RUNNING) { + return; + } + long commitOffsetFile = null != this.messageStore.getTimerMessageStore() ? this.messageStore.getTimerMessageStore().getCommitQueueOffset() : 0L; + long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); + long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB); + this.readOffset.set(maxCommitOffset); + this.expiredMessageReputService.start(); + this.rollMessageReputService.start(); + this.timeline.start(); + this.timerSysTopicScanService.start(); + this.state = RUNNING; + log.info("TimerMessageRocksDBStore start success, start commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, this.readOffset.get()); + } + + public synchronized boolean restart() { + try { + this.storeConfig.setTimerStopEnqueue(true); + if (this.state == RUNNING && !this.storeConfig.isTimerRocksDBStopScan()) { + log.info("restart TimerMessageRocksDBStore has been running"); + return true; Review Comment: I'm curious, if the enqueue operation was stopped using `setTimerStopEnqueue(true)` above, will there be any issues if we simply return here? ########## broker/src/main/java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBService.java: ########## @@ -0,0 +1,271 @@ +/* + * 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.transaction.rocksdb; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; +import java.util.concurrent.TimeUnit; +import io.netty.channel.Channel; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.common.ServiceThread; +import org.apache.rocketmq.common.ThreadFactoryImpl; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.utils.ThreadUtils; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; +import org.apache.rocketmq.remoting.protocol.header.CheckTransactionStateRequestHeader; +import org.apache.rocketmq.store.DefaultMessageStore; +import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage; +import org.apache.rocketmq.store.transaction.TransRocksDBRecord; +import org.apache.rocketmq.store.transaction.TransMessageRocksDBStore; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TRANS_COLUMN_FAMILY; + +public class TransactionalMessageRocksDBService { + private static final Logger log = LoggerFactory.getLogger(LoggerName.TRANSACTION_LOGGER_NAME); + private static final int MAX_BATCH_SIZE_FROM_ROCKSDB = 2000; + private static final int INITIAL = 0, RUNNING = 1, SHUTDOWN = 2; + private volatile int state = INITIAL; + + private final MessageRocksDBStorage messageRocksDBStorage; + private final TransMessageRocksDBStore transMessageRocksDBStore; + private final MessageStore messageStore; + private final BrokerController brokerController; + + private TransStatusCheckService transStatusService; + private ExecutorService checkTranStatusTaskExecutor; + + public TransactionalMessageRocksDBService(final MessageStore messageStore, final BrokerController brokerController) { + this.messageStore = messageStore; + this.transMessageRocksDBStore = messageStore.getTransRocksDBStore(); + this.messageRocksDBStorage = transMessageRocksDBStore.getMessageRocksDBStorage(); + this.brokerController = brokerController; + } + + public void start() { + if (this.state == RUNNING) { + return; + } + initService(); + this.transStatusService.start(); + this.state = RUNNING; + log.info("TransactionalMessageRocksDBService start success"); + } + + private void initService() { + this.transStatusService = new TransStatusCheckService(); + this.checkTranStatusTaskExecutor = ThreadUtils.newThreadPoolExecutor( + 2, + 5, + 100, + TimeUnit.SECONDS, + new ArrayBlockingQueue<>(2000), + new ThreadFactoryImpl("Transaction-rocksdb-msg-check-thread", brokerController.getBrokerIdentity()), + new CallerRunsPolicy()); + } + + public void shutdown() { + if (this.state != RUNNING || this.state == SHUTDOWN) { + return; + } + if (null != this.transStatusService) { + this.transStatusService.shutdown(); + } + if (null != this.checkTranStatusTaskExecutor) { + this.checkTranStatusTaskExecutor.shutdown(); + } + this.state = SHUTDOWN; + log.info("TransactionalMessageRocksDBService shutdown success"); + } + + private void checkTransStatus() { + long count = 0; + byte[] lastKey = null; + while (true) { + try { + List<TransRocksDBRecord> trs = messageRocksDBStorage.scanRecordsForTrans(TRANS_COLUMN_FAMILY, MAX_BATCH_SIZE_FROM_ROCKSDB, lastKey); + if (null == trs || CollectionUtils.isEmpty(trs)) { + log.info("TransactionalMessageRocksDBService checkTransStatus trs is empty"); + break; + } + count += trs.size(); + checkTransRecordsStatus(trs); + lastKey = trs.size() >= MAX_BATCH_SIZE_FROM_ROCKSDB ? trs.get(trs.size() - 1).getKeyBytes() : null; + if (null == lastKey) { + break; + } + } catch (Exception e) { + log.error("TransactionalMessageRocksDBService checkTransStatus error, error: {}, count: {}", e.getMessage(), count); + break; + } + } + log.info("TransactionalMessageRocksDBService checkTransStatus count: {}", count); + } + + private void checkTransRecordsStatus(List<TransRocksDBRecord> trs) { + if (CollectionUtils.isEmpty(trs)) { + log.error("TransactionalMessageRocksDBService checkTransRecordsStatus, trs is empty"); Review Comment: The `checkTransRecordsStatus` method is currently only called within the `checkTransStatus` method, and since the `checkTransStatus` method already checks for empty values, this method checks again for redundancy. Furthermore, the log level here is `ERROR`, while `checkTransStatus` uses `INFO`, creating inconsistencies. ########## store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java: ########## @@ -0,0 +1,619 @@ +/* + * 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.timer.rocksdb; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import com.conversantmedia.util.concurrent.DisruptorBlockingQueue; +import com.google.common.util.concurrent.RateLimiter; +import io.opentelemetry.api.common.Attributes; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.ServiceThread; +import org.apache.rocketmq.common.TopicFilterType; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.message.MessageAccessor; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageExtBrokerInner; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; +import org.apache.rocketmq.store.DefaultMessageStore; +import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.PutMessageResult; +import org.apache.rocketmq.store.StoreUtil; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsConstant; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsManager; +import org.apache.rocketmq.store.metrics.StoreMetricsManager; +import org.apache.rocketmq.store.queue.ConsumeQueueInterface; +import org.apache.rocketmq.store.queue.CqUnit; +import org.apache.rocketmq.store.queue.ReferredIterator; +import org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage; +import org.apache.rocketmq.store.stats.BrokerStatsManager; +import org.apache.rocketmq.store.timer.TimerMetrics; +import org.apache.rocketmq.store.util.PerfCounter; +import static org.apache.rocketmq.common.message.MessageConst.PROPERTY_TIMER_ROLL_LABEL; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMER_COLUMN_FAMILY; +import static org.apache.rocketmq.store.timer.TimerMessageStore.TIMER_TOPIC; + +public class TimerMessageRocksDBStore { + private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); + private static final Logger logError = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME); + private static final String SCAN_SYS_TOPIC = "scanSysTopic"; + private static final String SCAN_SYS_TOPIC_MISS = "scanSysTopicMiss"; + private static final String OUT_BIZ_MESSAGE = "outBizMsg"; + private static final String ROLL_LABEL = "R"; + private static final int PUT_OK = 0, PUT_NEED_RETRY = 1, PUT_NO_RETRY = 2; + private static final int MAX_GET_MSG_TIMES = 3, MAX_PUT_MSG_TIMES = 3; + private static final int TIME_UP_CAPACITY = 100, ROLL_CAPACITY = 50; + private static final int INITIAL = 0, RUNNING = 1, SHUTDOWN = 2; + private volatile int state = INITIAL; + private static long expirationThresholdMillis = 999L; + private final AtomicLong readOffset = new AtomicLong(0);//timerSysTopic read offset + private final MessageStore messageStore; + private final TimerMetrics timerMetrics; + private final MessageStoreConfig storeConfig; + private final BrokerStatsManager brokerStatsManager; + private final MessageRocksDBStorage messageRocksDBStorage; + private Timeline timeline; + private TimerSysTopicScanService timerSysTopicScanService; + private TimerMessageReputService expiredMessageReputService; + private TimerMessageReputService rollMessageReputService; + protected long precisionMs; + private BlockingQueue<List<TimerRocksDBRecord>> expiredMessageQueue; + private BlockingQueue<List<TimerRocksDBRecord>> rollMessageQueue; + protected final PerfCounter.Ticks perfCounterTicks = new PerfCounter.Ticks(log); + private Function<MessageExtBrokerInner, PutMessageResult> escapeBridgeHook; + private ThreadLocal<ByteBuffer> bufferLocal = null; + + public TimerMessageRocksDBStore(final MessageStore messageStore, final TimerMetrics timerMetrics, + final BrokerStatsManager brokerStatsManager) { + this.messageStore = messageStore; + this.storeConfig = messageStore.getMessageStoreConfig(); + this.precisionMs = storeConfig.getTimerRocksDBPrecisionMs() < 100L ? 1000L : storeConfig.getTimerRocksDBPrecisionMs(); + expirationThresholdMillis = precisionMs - 1L; + this.messageRocksDBStorage = messageStore.getMessageRocksDBStorage(); + this.timerMetrics = timerMetrics; + this.brokerStatsManager = brokerStatsManager; + bufferLocal = ThreadLocal.withInitial(() -> ByteBuffer.allocate(storeConfig.getMaxMessageSize())); + } + + public synchronized boolean load() { + initService(); + boolean result = this.timerMetrics.load(); + log.info("TimerMessageRocksDBStore load result: {}", result); + return result; + } + + public synchronized void start() { + if (this.state == RUNNING) { + return; + } + long commitOffsetFile = null != this.messageStore.getTimerMessageStore() ? this.messageStore.getTimerMessageStore().getCommitQueueOffset() : 0L; + long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); + long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB); + this.readOffset.set(maxCommitOffset); + this.expiredMessageReputService.start(); + this.rollMessageReputService.start(); + this.timeline.start(); + this.timerSysTopicScanService.start(); + this.state = RUNNING; + log.info("TimerMessageRocksDBStore start success, start commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, this.readOffset.get()); + } + + public synchronized boolean restart() { + try { + this.storeConfig.setTimerStopEnqueue(true); + if (this.state == RUNNING && !this.storeConfig.isTimerRocksDBStopScan()) { + log.info("restart TimerMessageRocksDBStore has been running"); + return true; + } + if (this.state == RUNNING && this.storeConfig.isTimerRocksDBStopScan()) { + long commitOffsetFile = null != this.messageStore.getTimerMessageStore() ? this.messageStore.getTimerMessageStore().getCommitQueueOffset() : 0L; + long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); + long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB); + this.readOffset.set(maxCommitOffset); + log.info("restart TimerMessageRocksDBStore has benn recover running, commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, readOffset.get()); + } else { + this.load(); Review Comment: If `load` fails here, should we continue with the subsequent `start`? ########## store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java: ########## @@ -0,0 +1,619 @@ +/* + * 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.timer.rocksdb; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import com.conversantmedia.util.concurrent.DisruptorBlockingQueue; +import com.google.common.util.concurrent.RateLimiter; +import io.opentelemetry.api.common.Attributes; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.ServiceThread; +import org.apache.rocketmq.common.TopicFilterType; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.message.MessageAccessor; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageExtBrokerInner; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; +import org.apache.rocketmq.store.DefaultMessageStore; +import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.PutMessageResult; +import org.apache.rocketmq.store.StoreUtil; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsConstant; +import org.apache.rocketmq.store.metrics.DefaultStoreMetricsManager; +import org.apache.rocketmq.store.metrics.StoreMetricsManager; +import org.apache.rocketmq.store.queue.ConsumeQueueInterface; +import org.apache.rocketmq.store.queue.CqUnit; +import org.apache.rocketmq.store.queue.ReferredIterator; +import org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage; +import org.apache.rocketmq.store.stats.BrokerStatsManager; +import org.apache.rocketmq.store.timer.TimerMetrics; +import org.apache.rocketmq.store.util.PerfCounter; +import static org.apache.rocketmq.common.message.MessageConst.PROPERTY_TIMER_ROLL_LABEL; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMER_COLUMN_FAMILY; +import static org.apache.rocketmq.store.timer.TimerMessageStore.TIMER_TOPIC; + +public class TimerMessageRocksDBStore { + private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); + private static final Logger logError = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME); + private static final String SCAN_SYS_TOPIC = "scanSysTopic"; + private static final String SCAN_SYS_TOPIC_MISS = "scanSysTopicMiss"; + private static final String OUT_BIZ_MESSAGE = "outBizMsg"; + private static final String ROLL_LABEL = "R"; + private static final int PUT_OK = 0, PUT_NEED_RETRY = 1, PUT_NO_RETRY = 2; + private static final int MAX_GET_MSG_TIMES = 3, MAX_PUT_MSG_TIMES = 3; + private static final int TIME_UP_CAPACITY = 100, ROLL_CAPACITY = 50; + private static final int INITIAL = 0, RUNNING = 1, SHUTDOWN = 2; + private volatile int state = INITIAL; + private static long expirationThresholdMillis = 999L; + private final AtomicLong readOffset = new AtomicLong(0);//timerSysTopic read offset + private final MessageStore messageStore; + private final TimerMetrics timerMetrics; + private final MessageStoreConfig storeConfig; + private final BrokerStatsManager brokerStatsManager; + private final MessageRocksDBStorage messageRocksDBStorage; + private Timeline timeline; + private TimerSysTopicScanService timerSysTopicScanService; + private TimerMessageReputService expiredMessageReputService; + private TimerMessageReputService rollMessageReputService; + protected long precisionMs; + private BlockingQueue<List<TimerRocksDBRecord>> expiredMessageQueue; + private BlockingQueue<List<TimerRocksDBRecord>> rollMessageQueue; + protected final PerfCounter.Ticks perfCounterTicks = new PerfCounter.Ticks(log); + private Function<MessageExtBrokerInner, PutMessageResult> escapeBridgeHook; + private ThreadLocal<ByteBuffer> bufferLocal = null; + + public TimerMessageRocksDBStore(final MessageStore messageStore, final TimerMetrics timerMetrics, + final BrokerStatsManager brokerStatsManager) { + this.messageStore = messageStore; + this.storeConfig = messageStore.getMessageStoreConfig(); + this.precisionMs = storeConfig.getTimerRocksDBPrecisionMs() < 100L ? 1000L : storeConfig.getTimerRocksDBPrecisionMs(); + expirationThresholdMillis = precisionMs - 1L; + this.messageRocksDBStorage = messageStore.getMessageRocksDBStorage(); + this.timerMetrics = timerMetrics; + this.brokerStatsManager = brokerStatsManager; + bufferLocal = ThreadLocal.withInitial(() -> ByteBuffer.allocate(storeConfig.getMaxMessageSize())); + } + + public synchronized boolean load() { + initService(); + boolean result = this.timerMetrics.load(); + log.info("TimerMessageRocksDBStore load result: {}", result); + return result; + } + + public synchronized void start() { + if (this.state == RUNNING) { + return; + } + long commitOffsetFile = null != this.messageStore.getTimerMessageStore() ? this.messageStore.getTimerMessageStore().getCommitQueueOffset() : 0L; + long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); + long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB); + this.readOffset.set(maxCommitOffset); + this.expiredMessageReputService.start(); + this.rollMessageReputService.start(); + this.timeline.start(); + this.timerSysTopicScanService.start(); + this.state = RUNNING; + log.info("TimerMessageRocksDBStore start success, start commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, this.readOffset.get()); + } + + public synchronized boolean restart() { + try { + this.storeConfig.setTimerStopEnqueue(true); + if (this.state == RUNNING && !this.storeConfig.isTimerRocksDBStopScan()) { + log.info("restart TimerMessageRocksDBStore has been running"); + return true; + } + if (this.state == RUNNING && this.storeConfig.isTimerRocksDBStopScan()) { + long commitOffsetFile = null != this.messageStore.getTimerMessageStore() ? this.messageStore.getTimerMessageStore().getCommitQueueOffset() : 0L; + long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); + long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB); + this.readOffset.set(maxCommitOffset); + log.info("restart TimerMessageRocksDBStore has benn recover running, commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, readOffset.get()); + } else { + this.load(); + this.start(); + } + this.storeConfig.setTimerRocksDBEnable(true); + this.storeConfig.setTimerRocksDBStopScan(false); + return true; + } catch (Exception e) { + logError.error("TimerMessageRocksDBStore restart error: {}", e.getMessage()); + return false; + } + } + + public void shutdown() { + if (this.state != RUNNING || this.state == SHUTDOWN) { Review Comment: ditto. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
