diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java index 07f16e98965..0b20050631d 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java @@ -568,7 +568,13 @@ private void reviveMsgFromCk(PopCheckPoint popCheckPoint) { // retry msg long msgOffset = popCheckPoint.ackOffsetByIndex((byte) j); CompletableFuture> future = getBizMessage(popCheckPoint, msgOffset) - .thenApply(rst -> { + .handle((rst, throwable) -> { + if (throwable != null) { + POP_LOGGER.error("reviveQueueId={}, get biz msg failed, topic:{}, qid:{}, offset:{}, brokerName:{}", + queueId, popCheckPoint.getTopic(), popCheckPoint.getQueueId(), msgOffset, + popCheckPoint.getBrokerName(), throwable); + return new Pair<>(msgOffset, false); + } MessageExt message = rst.getLeft(); if (message == null) { POP_LOGGER.info("reviveQueueId={}, can not get biz msg, topic:{}, qid:{}, offset:{}, brokerName:{}, info:{}, retry:{}, then continue", diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java index fa7e9982e1f..cbd702f9e8a 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java @@ -17,6 +17,7 @@ package org.apache.rocketmq.broker.processor; import com.alibaba.fastjson2.JSON; +import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.commons.lang3.tuple.Triple; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.failover.EscapeBridge; @@ -58,6 +59,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; +import java.util.NavigableMap; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; @@ -389,6 +391,33 @@ public void testReviveMsgFromCk_messageNotFound_needRetry() throws Throwable { verify(messageStore, times(1)).putMessage(any(MessageExtBrokerInner.class)); // rewrite CK } + @Test + public void testReviveMsgFromCk_getBizMessageExceptional_rewriteCK() throws Throwable { + PopCheckPoint ck = buildPopCheckPoint(0, 0, 1); + PopReviveService.ConsumeReviveObj reviveObj = new PopReviveService.ConsumeReviveObj(); + reviveObj.map.put("", ck); + reviveObj.endTime = System.currentTimeMillis(); + + ArgumentCaptor commitOffsetCaptor = ArgumentCaptor.forClass(Long.class); + doNothing().when(consumerOffsetManager).commitOffset(anyString(), anyString(), anyString(), anyInt(), + commitOffsetCaptor.capture()); + + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new RuntimeException("store read failed")); + when(escapeBridge.getMessageAsync(anyString(), anyLong(), anyInt(), anyString(), anyBoolean())) + .thenReturn(failed); + + popReviveService.mergeAndRevive(reviveObj); + + NavigableMap inflight = (NavigableMap) FieldUtils.readField( + popReviveService, "inflightReviveRequestMap", true); + assertEquals(1, reviveObj.newOffset); + assertEquals(1, commitOffsetCaptor.getValue().longValue()); + assertEquals(0, inflight.size()); + // An exceptional async read must retain retryability by rewriting the checkpoint. + verify(messageStore, times(1)).putMessage(any(MessageExtBrokerInner.class)); + } + @Test public void testReviveMsgFromCk_messageNotFound_needRetry_end() throws Throwable { brokerConfig.setSkipWhenCKRePutReachMaxTimes(true);