From 116dd69ec59e7b5456e6adc698cfc91c38a40354 Mon Sep 17 00:00:00 2001 From: "wangjiahua.wjh" Date: Sat, 18 Jul 2026 23:29:22 +0800 Subject: [PATCH 1/3] [ISSUE #10658] Fix PopConsumerService.revive aborting the whole batch on a single failed record revive(PopConsumerRecord) chains getMessageAsync(record).thenCompose(...) with no exceptionally handler, and the batch revive(AtomicLong, int) awaits all per-record futures via CompletableFuture.allOf(...).join() with no surrounding try-catch. As a result a single failing record aborts the entire batch: writeRecords(failureList), deleteRecords(consumerRecords) and the currentTime advance are all skipped and revive() throws. If one record keeps failing (e.g. a persistently unreachable remote), revive gets stuck reprocessing the same batch forever, and the retries for the other healthy records in that batch are never persisted (head-of-line blocking). A record can fail in two ways: (1) the returned future completes exceptionally (e.g. reviveRetry throwing inside thenCompose, or a decode failure in the escape bridge), and (2) revive(record) throws synchronously before it returns a future (e.g. DefaultMessageStore.getMessageAsync is completedFuture(getMessage(...)) and getMessage throws). Handle both: add an exceptionally handler to revive(record) that logs and downgrades an async failure to a failed revive (returns false); and in the batch loop, catch a synchronous throw from revive(record) and turn it into completedFuture(false) instead of rethrowing and aborting the batch (the semaphore permit is released by the whenComplete stage, and acquire()'s InterruptedException is handled separately). Either way a single record's failure only affects that record, and writeRecords/deleteRecords/ currentTime still run. Add PopConsumerServiceTest#reviveShouldNotAbortBatchWhenGetMessageFailsExceptionally (async exceptional completion) and reviveShouldIsolateSynchronousReadFailureAndNotBlockHealthyRecords (synchronous throw plus head-of-line blocking); both fail before the fix and pass after. --- .../broker/pop/PopConsumerService.java | 22 ++++- .../broker/pop/PopConsumerServiceTest.java | 97 +++++++++++++++++++ 2 files changed, 116 insertions(+), 3 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java index f72e2ba26f2..a3f74ad6a0c 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java @@ -587,6 +587,13 @@ public CompletableFuture revive(PopConsumerRecord record) { return CompletableFuture.completedFuture(!result.getRight()); } return CompletableFuture.completedFuture(this.reviveRetry(record, result.getLeft())); + }) + .exceptionally(throwable -> { + // Do not let a single failed async read (e.g. a remote read via the escape bridge) + // abort the whole revive batch in revive(AtomicLong, int). Treat it as a failed + // revive so the record is scheduled for retry instead of throwing out of allOf().join(). + log.error("PopConsumerService revive failed, will retry, record={}", record, throwable); + return false; }); } @@ -613,13 +620,22 @@ public long revive(AtomicLong currentTime, int maxCount) { // could merge read operation here for (PopConsumerRecord record : consumerRecords) { - CompletableFuture future; try { semaphore.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + CompletableFuture future; + try { future = this.revive(record); } catch (Exception e) { - semaphore.release(); - throw new RuntimeException(e); + // A synchronous failure from revive(record) (e.g. getMessageAsync throwing before it + // returns a future) must not abort the whole batch; treat it as a failed revive so the + // record goes through the failureList backoff-retry path below. The semaphore permit is + // released by the whenComplete stage attached to this future. + log.error("PopConsumerService revive threw synchronously, will retry, record={}", record, e); + future = CompletableFuture.completedFuture(false); } futureList.add(future.thenAccept(result -> { if (!result) { diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java index 44189744b46..5421fd0af4d 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java @@ -743,4 +743,101 @@ public void testReviveRetryWithSuspendFalseMultipleTimes() { messageExt.setReconsumeTimes(capturedMessage.getReconsumeTimes()); } } + + @Test + public void reviveShouldNotAbortBatchWhenGetMessageFailsExceptionally() { + // A single getMessageAsync that completes exceptionally (e.g. a transient remote read + // failure via the escape bridge) must not abort the whole revive batch. Before the fix, + // revive(record) had no exceptionally handler, so the exception propagated through + // allOf(...).join(), skipping writeRecords/deleteRecords and throwing out of revive(). + Mockito.when(brokerController.getEscapeBridge()).thenReturn(Mockito.mock(EscapeBridge.class)); + Mockito.when(brokerController.getSubscriptionGroupManager() + .containsSubscriptionGroup(anyString())).thenReturn(true); + PopConsumerService consumerServiceSpy = Mockito.spy(consumerService); + + consumerService.getPopConsumerStore().start(); + + long popTime = 1000000000L; + long invisibleTime = 60 * 1000L; + PopConsumerRecord record = new PopConsumerRecord(); + record.setPopTime(popTime); + record.setInvisibleTime(invisibleTime); + record.setTopicId("topic"); + record.setGroupId("group"); + record.setQueueId(0); + record.setOffset(0); + consumerService.getPopConsumerStore().writeRecords(Collections.singletonList(record)); + + // getMessageAsync completes exceptionally to simulate a remote read failure. + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new RuntimeException("simulated remote read failure")); + Mockito.doReturn(failed).when(consumerServiceSpy).getMessageAsync(any(PopConsumerRecord.class)); + + long visibleTimestamp = popTime + invisibleTime; + + // revive must not throw, and the batch bookkeeping must still run: the record is consumed + // from the scan window (deleteRecords ran) instead of the whole batch being aborted. + Assert.assertEquals(1, consumerServiceSpy.revive(new AtomicLong(visibleTimestamp), 1)); + Assert.assertEquals(0, consumerService.getPopConsumerStore() + .scanExpiredRecords(0, visibleTimestamp, 1).size()); + + consumerService.shutdown(); + } + + @Test + public void reviveShouldIsolateSynchronousReadFailureAndNotBlockHealthyRecords() { + // A record whose read fails *synchronously* (getMessageAsync throwing before it returns a + // future, e.g. DefaultMessageStore.getMessage throwing inside completedFuture(getMessage(...))) + // must not abort the whole revive batch. The per-record .exceptionally handler does NOT cover + // this path, because the throw happens before the thenCompose/exceptionally chain is attached; + // the batch loop caught it and rethrew as RuntimeException, skipping writeRecords/deleteRecords + // and blocking the healthy records in the same batch (head-of-line blocking). + Mockito.when(brokerController.getEscapeBridge()).thenReturn(Mockito.mock(EscapeBridge.class)); + Mockito.when(brokerController.getSubscriptionGroupManager() + .containsSubscriptionGroup(anyString())).thenReturn(true); + PopConsumerService consumerServiceSpy = Mockito.spy(consumerService); + + consumerService.getPopConsumerStore().start(); + + long popTime = 1000000000L; + long invisibleTime = 60 * 1000L; + + // record at offset 0 -> read fails synchronously; record at offset 1 -> healthy. + PopConsumerRecord bad = new PopConsumerRecord(); + bad.setPopTime(popTime); + bad.setInvisibleTime(invisibleTime); + bad.setTopicId("topic"); + bad.setGroupId("group"); + bad.setQueueId(0); + bad.setOffset(0); + PopConsumerRecord healthy = new PopConsumerRecord(); + healthy.setPopTime(popTime); + healthy.setInvisibleTime(invisibleTime); + healthy.setTopicId("topic"); + healthy.setGroupId("group"); + healthy.setQueueId(0); + healthy.setOffset(1); + consumerService.getPopConsumerStore().writeRecords(Collections.singletonList(bad)); + consumerService.getPopConsumerStore().writeRecords(Collections.singletonList(healthy)); + + // bad record: getMessageAsync throws synchronously (not an exceptionally-completed future). + Mockito.doThrow(new RuntimeException("simulated synchronous read failure")) + .when(consumerServiceSpy).getMessageAsync(Mockito.argThat( + (PopConsumerRecord r) -> r != null && r.getOffset() == 0L)); + // healthy record: read returns no message needing retry -> revive succeeds. + Mockito.doReturn(CompletableFuture.completedFuture(Triple.of((MessageExt) null, "", false))) + .when(consumerServiceSpy).getMessageAsync(Mockito.argThat( + (PopConsumerRecord r) -> r != null && r.getOffset() == 1L)); + + long visibleTimestamp = popTime + invisibleTime; + + // revive must process the whole batch (returns 2) without throwing, and both original records + // must be consumed from the scan window (deleteRecords ran) - proving the failed record was + // isolated and the healthy record was not blocked. + Assert.assertEquals(2, consumerServiceSpy.revive(new AtomicLong(visibleTimestamp), 10)); + Assert.assertEquals(0, consumerService.getPopConsumerStore() + .scanExpiredRecords(0, visibleTimestamp, 10).size()); + + consumerService.shutdown(); + } } \ No newline at end of file From 9f38ed5cb5f818972ce7385706d9f37ee20fc60a Mon Sep 17 00:00:00 2001 From: "wangjiahua.wjh" Date: Tue, 28 Jul 2026 15:59:59 +0800 Subject: [PATCH 2/3] [ISSUE #10658] Attach the exception-to-false conversion at the batch call site only Address the review on #10659: revive(PopConsumerRecord) is also used as the Consumer callback of PopConsumerCache (enablePopBufferMerge), where the returned future is discarded, so a false result would never be consumed there. Converting exceptions to false inside revive(record) therefore created a contract that one caller silently ignores, and the "will retry" log would be misleading on that path. Move the .exceptionally handler out of revive(record) to the batch call site in revive(AtomicLong, int), where a false result is actually consumed and turned into a failureList backoff retry. revive(record) keeps its original exception semantics, so the PopConsumerCache path behaves exactly as it did before this PR. The pre-existing issue that an asynchronously failed revive can be cleared from the cache without being persisted for retry is tracked separately. Both regression tests still pass: reviveShouldNotAbortBatchWhenGetMessageFailsExceptionally and reviveShouldIsolateSynchronousReadFailureAndNotBlockHealthyRecords. --- .../rocketmq/broker/pop/PopConsumerService.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java index a3f74ad6a0c..ded2db2f8e4 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java @@ -587,13 +587,6 @@ public CompletableFuture revive(PopConsumerRecord record) { return CompletableFuture.completedFuture(!result.getRight()); } return CompletableFuture.completedFuture(this.reviveRetry(record, result.getLeft())); - }) - .exceptionally(throwable -> { - // Do not let a single failed async read (e.g. a remote read via the escape bridge) - // abort the whole revive batch in revive(AtomicLong, int). Treat it as a failed - // revive so the record is scheduled for retry instead of throwing out of allOf().join(). - log.error("PopConsumerService revive failed, will retry, record={}", record, throwable); - return false; }); } @@ -628,7 +621,14 @@ public long revive(AtomicLong currentTime, int maxCount) { } CompletableFuture future; try { - future = this.revive(record); + // Attach the exception-to-false conversion at this call site only: here a false + // result is consumed below and turned into a failureList backoff retry. Other + // callers of revive(record) (e.g. the PopConsumerCache callback) do not consume + // the result and must keep the original exception semantics. + future = this.revive(record).exceptionally(throwable -> { + log.error("PopConsumerService revive failed, will retry, record={}", record, throwable); + return false; + }); } catch (Exception e) { // A synchronous failure from revive(record) (e.g. getMessageAsync throwing before it // returns a future) must not abort the whole batch; treat it as a failed revive so the From 117c753ca335f3646f46ebab3df3f1c8f1fd19c9 Mon Sep 17 00:00:00 2001 From: "wangjiahua.wjh" Date: Wed, 29 Jul 2026 09:24:09 +0800 Subject: [PATCH 3/3] [ISSUE #10658] Consume incidental interrupts and preserve the suspend flag in backoff records Address two review comments on #10659. Interrupt handling: restoring the interrupt flag before rethrowing left the revive service permanently interrupted, because shutdown stops this service with the stopped flag and wakeup() rather than interruption, and ServiceThread.waitForRunning preserves the interrupt status. Every later Semaphore.acquire() threw again, turning one incidental interrupt into a permanent busy error loop with each batch abandoned before writeRecords and deleteRecords. Consume incidental interrupts and retry the acquire; if the service is stopped, abort the batch cleanly so the records are reprocessed after the next start. Suspend flag: the backoff record was built with the eight-argument PopConsumerRecord constructor, which defaults suspend to false. A suspended record whose revive failed transiently lost the flag, so a later successful reviveRetry incremented reconsumeTimes unexpectedly. Build the retry record with the nine-argument constructor and preserve record.isSuspend(). New regression tests: one blocks the worker on the semaphore with a single permit, interrupts it once, releases the outstanding read and verifies the whole batch still completes without an exception; the other fails one record synchronously and one asynchronously, both with suspend=true, and verifies the persisted backoff records keep suspend=true with attemptTimes=1. Both fail before this change. --- .../broker/pop/PopConsumerService.java | 26 +++- .../broker/pop/PopConsumerServiceTest.java | 120 ++++++++++++++++++ 2 files changed, 139 insertions(+), 7 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java index ded2db2f8e4..72a44d79c2b 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java @@ -613,11 +613,23 @@ public long revive(AtomicLong currentTime, int maxCount) { // could merge read operation here for (PopConsumerRecord record : consumerRecords) { - try { - semaphore.acquire(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RuntimeException(e); + while (true) { + try { + semaphore.acquire(); + break; + } catch (InterruptedException e) { + // Shutdown stops this service with the stopped flag and wakeup() rather than + // thread interruption, and ServiceThread.waitForRunning preserves the interrupt + // status. Restoring the flag here would make every later acquire() throw again + // and leave the revive service in a permanent busy loop, so consume incidental + // interrupts and retry. If the service is being stopped (e.g. shutdown(true)), + // abort the batch cleanly: the records are not yet deleted from the store and + // will be reprocessed after the next start. + if (this.isStopped()) { + throw new RuntimeException("PopConsumerService stopped while acquiring the revive semaphore", e); + } + log.warn("PopConsumerService interrupted while acquiring the revive semaphore, retry"); + } } CompletableFuture future; try { @@ -644,8 +656,8 @@ public long revive(AtomicLong currentTime, int maxCount) { Math.min(REWRITE_INTERVALS_IN_SECONDS.length - 1, record.getAttemptTimes())]; long nextInvisibleTime = record.getInvisibleTime() + backoffInterval; PopConsumerRecord retryRecord = new PopConsumerRecord(System.currentTimeMillis(), - record.getGroupId(), record.getTopicId(), record.getQueueId(), - record.getRetryFlag(), nextInvisibleTime, record.getOffset(), record.getAttemptId()); + record.getGroupId(), record.getTopicId(), record.getQueueId(), record.getRetryFlag(), + nextInvisibleTime, record.getOffset(), record.getAttemptId(), record.isSuspend()); retryRecord.setAttemptTimes(record.getAttemptTimes() + 1); failureList.add(retryRecord); log.warn("PopConsumerService revive backoff retry, record={}", retryRecord); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java index 5421fd0af4d..ddac6f58c68 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java @@ -24,8 +24,10 @@ import java.util.List; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import java.util.stream.IntStream; import org.apache.commons.io.FileUtils; @@ -840,4 +842,122 @@ public void reviveShouldIsolateSynchronousReadFailureAndNotBlockHealthyRecords() consumerService.shutdown(); } + + @Test + public void reviveShouldSurviveIncidentalInterruptWhileAcquiringSemaphore() throws Exception { + // An incidental interrupt while waiting on the per-batch semaphore used to restore the + // interrupt flag and abort the batch. Because shutdown stops this service with the + // stopped flag rather than interruption, and ServiceThread.waitForRunning preserves the + // interrupt status, the restored flag made every later acquire() throw again and left + // the revive service in a permanent busy loop. An incidental interrupt must be consumed + // and the batch must finish normally. + Mockito.when(brokerController.getEscapeBridge()).thenReturn(Mockito.mock(EscapeBridge.class)); + Mockito.when(brokerController.getSubscriptionGroupManager() + .containsSubscriptionGroup(anyString())).thenReturn(true); + brokerController.getBrokerConfig().setPopReviveConcurrency(1); + PopConsumerService consumerServiceSpy = Mockito.spy(consumerService); + + consumerService.getPopConsumerStore().start(); + + long popTime = 1000000000L; + long invisibleTime = 60 * 1000L; + PopConsumerRecord first = new PopConsumerRecord(popTime, "group", "topic", 0, + PopConsumerRecord.RetryType.NORMAL_TOPIC.getCode(), invisibleTime, 0, attemptId); + PopConsumerRecord second = new PopConsumerRecord(popTime, "group", "topic", 0, + PopConsumerRecord.RetryType.NORMAL_TOPIC.getCode(), invisibleTime, 1, attemptId); + consumerService.getPopConsumerStore().writeRecords(Collections.singletonList(first)); + consumerService.getPopConsumerStore().writeRecords(Collections.singletonList(second)); + + // the first read never completes until released, so the single permit stays taken and + // the worker blocks inside semaphore.acquire() for the second record + CompletableFuture> gate = new CompletableFuture<>(); + CountDownLatch firstReadStarted = new CountDownLatch(1); + Mockito.doAnswer(invocation -> { + firstReadStarted.countDown(); + return gate; + }).when(consumerServiceSpy).getMessageAsync(Mockito.argThat( + (PopConsumerRecord r) -> r != null && r.getOffset() == 0L)); + Mockito.doReturn(CompletableFuture.completedFuture(Triple.of((MessageExt) null, "", false))) + .when(consumerServiceSpy).getMessageAsync(Mockito.argThat( + (PopConsumerRecord r) -> r != null && r.getOffset() == 1L)); + + long visibleTimestamp = popTime + invisibleTime; + AtomicLong reviveCount = new AtomicLong(-1); + AtomicReference thrown = new AtomicReference<>(); + Thread worker = new Thread(() -> { + try { + reviveCount.set(consumerServiceSpy.revive(new AtomicLong(visibleTimestamp), 10)); + } catch (Throwable t) { + thrown.set(t); + } + }, "revive-interrupt-test"); + worker.start(); + + Assert.assertTrue(firstReadStarted.await(5, TimeUnit.SECONDS)); + long deadline = System.currentTimeMillis() + 5000; + while (worker.getState() != Thread.State.WAITING && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + Assert.assertEquals(Thread.State.WAITING, worker.getState()); + + // the incidental interrupt, then release the outstanding operation + worker.interrupt(); + Thread.sleep(100); + gate.complete(Triple.of((MessageExt) null, "", false)); + + worker.join(TimeUnit.SECONDS.toMillis(10)); + Assert.assertFalse(worker.isAlive()); + Assert.assertNull("revive must not abort on an incidental interrupt", thrown.get()); + Assert.assertEquals(2, reviveCount.get()); + Assert.assertEquals(0, consumerService.getPopConsumerStore() + .scanExpiredRecords(0, visibleTimestamp, 10).size()); + + consumerService.shutdown(); + } + + @Test + public void reviveBackoffRecordShouldPreserveSuspendFlag() { + // When a revive failure is converted to false, the backoff record was built with the + // eight-argument PopConsumerRecord constructor, which defaults suspend to false. For a + // record created with suspend=true (changeInvisibilityDuration), losing the flag makes a + // later successful reviveRetry increment reconsumeTimes although the failure was only + // transient. Both the synchronous and the asynchronous failure paths must preserve it. + Mockito.when(brokerController.getEscapeBridge()).thenReturn(Mockito.mock(EscapeBridge.class)); + Mockito.when(brokerController.getSubscriptionGroupManager() + .containsSubscriptionGroup(anyString())).thenReturn(true); + PopConsumerService consumerServiceSpy = Mockito.spy(consumerService); + + consumerService.getPopConsumerStore().start(); + + long popTime = 1000000000L; + long invisibleTime = 60 * 1000L; + PopConsumerRecord asyncFail = new PopConsumerRecord(popTime, "group", "topic", 0, + PopConsumerRecord.RetryType.NORMAL_TOPIC.getCode(), invisibleTime, 0, attemptId, true); + PopConsumerRecord syncFail = new PopConsumerRecord(popTime, "group", "topic", 0, + PopConsumerRecord.RetryType.NORMAL_TOPIC.getCode(), invisibleTime, 1, attemptId, true); + consumerService.getPopConsumerStore().writeRecords(Collections.singletonList(asyncFail)); + consumerService.getPopConsumerStore().writeRecords(Collections.singletonList(syncFail)); + + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new RuntimeException("simulated async read failure")); + Mockito.doReturn(failed).when(consumerServiceSpy).getMessageAsync(Mockito.argThat( + (PopConsumerRecord r) -> r != null && r.getOffset() == 0L)); + Mockito.doThrow(new RuntimeException("simulated sync read failure")) + .when(consumerServiceSpy).getMessageAsync(Mockito.argThat( + (PopConsumerRecord r) -> r != null && r.getOffset() == 1L)); + + long visibleTimestamp = popTime + invisibleTime; + Assert.assertEquals(2, consumerServiceSpy.revive(new AtomicLong(visibleTimestamp), 10)); + + // the originals are consumed and replaced by backoff records that keep suspend=true + List retryRecords = consumerService.getPopConsumerStore().scanExpiredRecords( + 0, System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1), 10); + Assert.assertEquals(2, retryRecords.size()); + for (PopConsumerRecord retryRecord : retryRecords) { + Assert.assertEquals(1, retryRecord.getAttemptTimes()); + Assert.assertTrue("suspend flag must be preserved in the backoff record", retryRecord.isSuspend()); + } + + consumerService.shutdown(); + } } \ No newline at end of file