From 831fcc76cd7cd362bb6c136c287c624bb7eaf40a Mon Sep 17 00:00:00 2001 From: lizhimins <707364882@qq.com> Date: Tue, 19 Sep 2023 10:04:04 +0800 Subject: [PATCH] [ISSUE #7363] Fix get message from tiered storage return incorrect next pull offset (#7365) --- .../tieredstore/TieredMessageFetcher.java | 2 +- .../tieredstore/TieredMessageStore.java | 29 ++++++++++--------- .../tieredstore/TieredMessageStoreTest.java | 5 ++-- 3 files changed, 20 insertions(+), 16 deletions(-) diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageFetcher.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageFetcher.java index 766ff64f6c..c948fa3fa1 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageFetcher.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageFetcher.java @@ -319,7 +319,7 @@ public class TieredMessageFetcher implements MessageStoreFetcher { } // if cache is miss, immediately pull messages - LOGGER.warn("TieredMessageFetcher#getMessageFromCacheAsync: cache miss: " + + LOGGER.info("TieredMessageFetcher#getMessageFromCacheAsync: cache miss: " + "topic: {}, queue: {}, queue offset: {}, max message num: {}", mq.getTopic(), mq.getQueueId(), queueOffset, maxCount); diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java index 9fb1b2f01c..d7d13d61e2 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java @@ -147,6 +147,11 @@ public class TieredMessageStore extends AbstractPluginMessageStore { public CompletableFuture getMessageAsync(String group, String topic, int queueId, long offset, int maxMsgNums, MessageFilter messageFilter) { + // For system topic, force reading from local store + if (TieredStoreUtil.isSystemTopic(topic) || PopAckConstants.isStartWithRevivePrefix(topic)) { + return next.getMessageAsync(group, topic, queueId, offset, maxMsgNums, messageFilter); + } + if (fetchFromCurrentStore(topic, queueId, offset, maxMsgNums)) { logger.trace("GetMessageAsync from current store, topic: {}, queue: {}, offset: {}", topic, queueId, offset); } else { @@ -158,6 +163,7 @@ public class TieredMessageStore extends AbstractPluginMessageStore { return fetcher .getMessageAsync(group, topic, queueId, offset, maxMsgNums, messageFilter) .thenApply(result -> { + Attributes latencyAttributes = TieredStoreMetricsManager.newAttributesBuilder() .put(TieredStoreMetricsConstant.LABEL_OPERATION, TieredStoreMetricsConstant.OPERATION_API_GET_MESSAGE) .put(TieredStoreMetricsConstant.LABEL_TOPIC, topic) @@ -166,8 +172,7 @@ public class TieredMessageStore extends AbstractPluginMessageStore { TieredStoreMetricsManager.apiLatency.record(stopwatch.elapsed(TimeUnit.MILLISECONDS), latencyAttributes); if (result.getStatus() == GetMessageStatus.OFFSET_FOUND_NULL || - result.getStatus() == GetMessageStatus.OFFSET_OVERFLOW_ONE || - result.getStatus() == GetMessageStatus.OFFSET_OVERFLOW_BADLY) { + result.getStatus() == GetMessageStatus.NO_MATCHED_LOGIC_QUEUE) { if (next.checkInStoreByConsumeOffset(topic, queueId, offset)) { TieredStoreMetricsManager.fallbackTotal.add(1, latencyAttributes); @@ -178,14 +183,8 @@ public class TieredMessageStore extends AbstractPluginMessageStore { } } - // Fetch system topic data from the broker when using the force level. - if (result.getStatus() == GetMessageStatus.NO_MATCHED_LOGIC_QUEUE) { - if (TieredStoreUtil.isSystemTopic(topic) || PopAckConstants.isStartWithRevivePrefix(topic)) { - return next.getMessage(group, topic, queueId, offset, maxMsgNums, messageFilter); - } - } - if (result.getStatus() != GetMessageStatus.FOUND && + result.getStatus() != GetMessageStatus.NO_MATCHED_LOGIC_QUEUE && result.getStatus() != GetMessageStatus.OFFSET_OVERFLOW_ONE && result.getStatus() != GetMessageStatus.OFFSET_OVERFLOW_BADLY) { logger.warn("GetMessageAsync not found and message is not in next store, result: {}, " + @@ -206,10 +205,14 @@ public class TieredMessageStore extends AbstractPluginMessageStore { if (minOffsetInQueue >= 0 && minOffsetInQueue < result.getMinOffset()) { result.setMinOffset(minOffsetInQueue); } - long maxOffsetInQueue = next.getMaxOffsetInQueue(topic, queueId); - if (maxOffsetInQueue >= 0 && maxOffsetInQueue > result.getMaxOffset()) { - result.setMaxOffset(maxOffsetInQueue); - } + + // In general, the local cq offset is slightly greater than the commit offset in read message, + // so there is no need to update the maximum offset to the local cq offset here, + // otherwise it will cause repeated consumption after next begin offset over commit offset. + + logger.trace("GetMessageAsync result, group: {}, topic: {}, queueId: {}, offset: {}, count:{}, {}", + group, topic, queueId, offset, maxMsgNums, result); + return result; }).exceptionally(e -> { logger.error("GetMessageAsync from tiered store failed", e); diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/TieredMessageStoreTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/TieredMessageStoreTest.java index 2451199c28..07af1fc8b1 100644 --- a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/TieredMessageStoreTest.java +++ b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/TieredMessageStoreTest.java @@ -168,7 +168,7 @@ public class TieredMessageStoreTest { GetMessageResult result1 = new GetMessageResult(); result1.setStatus(GetMessageStatus.FOUND); GetMessageResult result2 = new GetMessageResult(); - result2.setStatus(GetMessageStatus.MESSAGE_WAS_REMOVING); + result2.setStatus(GetMessageStatus.OFFSET_OVERFLOW_BADLY); when(fetcher.getMessageAsync(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any())).thenReturn(CompletableFuture.completedFuture(result1)); when(nextStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any())).thenReturn(result2); @@ -188,7 +188,8 @@ public class TieredMessageStoreTest { properties.setProperty("tieredStorageLevel", "3"); configuration.update(properties); when(nextStore.checkInStoreByConsumeOffset(anyString(), anyInt(), anyLong())).thenReturn(true); - Assert.assertSame(result2, store.getMessage("group", mq.getTopic(), mq.getQueueId(), 0, 0, null)); + Assert.assertEquals(result2.getStatus(), + store.getMessage("group", mq.getTopic(), mq.getQueueId(), 0, 0, null).getStatus()); } @Test