[improve][pip] PIP-423: Add a new admin API to acknowledge a single message #24370

Denovo1998 · 2025-06-01T13:13:57Z

Motivation

Operators and SREs occasionally need to intervene in a topic's backlog to handle problematic messages or adapt to changing business requirements. For instance:

Cancelling Scheduled Actions: A delayed message representing a future task (e.g., a scheduled report or a notification) may become obsolete. The most efficient way to handle this is to prevent its delivery entirely by acknowledging it pre-emptively.
Removing Backlogs: A specific message in a backlog might have a malformed payload that causes consumer applications to crash repeatedly. Removing this single "poison pill" message without affecting valid messages around it is a critical operational capability. This also applies to removing a single bad message from within a larger batch.
Manual Business Logic Correction: An event may have been sent that is later determined to be invalid due to external factors. An administrator needs a precise tool to remove this specific event from a subscription's delivery queue.

The existing skip(numMessages) API is a blunt instrument, ill-suited for these precise, targeted operations. This proposal introduces an administrative API to skip messages by their specific MessageId (including ledgerId, entryId, and optional batchIndex), providing a robust and reliable way to remove any individual message—delayed or not—from a subscription's backlog.

Implementation PR: #23907

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: Denovo1998#8

… Messages

Denovo1998 · 2025-06-03T13:35:35Z

Currently, there are some issues with this implementation.
(Note that this PIP only discusses the implementation of BucketDelayedDeliveryTracker)

If a delayed message in the LastMutableBucket has not been flushed to Bookie yet, and a failure occurs, the data in the LastMutableBucket will be lost. However, this will not have any impact, as after restarting, messages will still be read from the MackDelete position and the Bucket will be rebuilt.

However, our delayed message deletion command is different. If there is a LastMutableBucket that has not been successfully persisted (sealBucketAndAsyncPersistent), and the Broker crashes, this command will be lost. We cannot wait for the Seal Bucket condition to be triggered before returning the success of canceling the delayed message command, because we do not know how long it will take.

This part is important and needs to be resolved. I will think about how to solve this part, and everyone is also welcome to discuss with me.

I have some solutions, let me try testing them out first.

…ge message.

Denovo1998 · 2025-06-08T05:44:56Z

The document has been updated, we will implement the delayed message cancellation function through acknowledge message.

…f delayed messages.

pip/pip-423.md

lhotari · 2025-10-13T07:42:04Z

pip/pip-423.md

+*   **HTTP Body Parameters (JSON):** A map where keys are `ledgerId` as a string and values are `entryId` as a string.
+    *   **Example Body:** `{"12345": "100", "12346": "200"}`


The serialization format should be changed. The problem with this format is that it cannot be extended to cover batch messages.

besides ledgerId and entryId, there's also batchIndex, which is part of the message id.

One of the design principles has been to hide the implementation details of the message id in Pulsar and not leak it. That's why there are org.apache.pulsar.client.api.MessageId#toByteArray and org.apache.pulsar.client.api.MessageId#fromByteArray .

It would be useful to support multiple formats for the message id. For consistency with the MessageId.toByteArray() API, there should be support for accepting a list of base64 encoded byte array representations of the MessageId.
One way to solve this is to have the body map to an object that is mapped with Jackson using polymorphism configured with @JsonTypeInfo and @JsonSubTypes annotations. That would allow flexibility for adding various formats for the message id list.

the default format could be the byteArrays encoded as base64:

{"type": "byteArray", "messageIds": ["CLlgEAQwAA==", "CLlgEAYwAA==", "CLlgEJ4BMAA=", "CLlgEKQCMAA=", "CLlgEPQCMAA=", "CLlgECcwAA==", "CLlgEIQDMAA=", "CLlgEKEDMAA=", "CLlgEAQwAA==", "CLlgEI4EMAA="]}

since this could be made the default, the "type": "byteArray" could be omitted:

{"messageIds": ["CLlgEAQwAA==", "CLlgEAYwAA==", "CLlgEJ4BMAA=", "CLlgEKQCMAA=", "CLlgEPQCMAA=", "CLlgECcwAA==", "CLlgEIQDMAA=", "CLlgEKEDMAA=", "CLlgEAQwAA==", "CLlgEI4EMAA="]}

another could be the ledgerId:entryId pairs (without support for batchIndex) if this is absolutely necessary:

{"type": "map_of_ledgerId_entryId", "messageIds": {12345: 100, 12346: 200}}

Instead of that, it could be useful to have a format that maps directly:

{"type": "messageId", "messageIds": [{"ledgerId": 12345, "entryId": 100}, {"ledgerId": 12346, "entryId": 200}]}

this format could also support the batch index.

{"type": "messageId", "messageIds": [{"ledgerId": 12345, "entryId": 100, "batchIndex": 5}, {"ledgerId": 12346, "entryId": 200, "batchIndex": 10}]}

lhotari · 2025-10-13T07:51:55Z

pip/pip-423.md

+*   **Usage:** `pulsar-admin topics skip-messages <topic-name> [options]`
+*   **Options:**
+    *   `-s, --subscription <subName>` (required): The subscription to skip messages on.
+    *   `-m, --messageId <ledgerId=entryId>` (required): The message ID to skip. This option can be specified multiple times.


it should be possible to also specify the batch index.
A better format is ledgerId:entryId[:batchIndex] where the separator is : and the batch index is optional.

lhotari · 2025-10-13T07:53:39Z

pip/pip-423.md

+    // ... existing methods
+    CompletableFuture<Void> skipMessages(int numMessagesToSkip);
+
+    CompletableFuture<Void> skipMessages(Map<String, String> messageIds);


should be List<MessageIdAdv> in this interface.

I'm not actually sure if List<MessageIdAdv> is the best option here. I guess we'd need a new type (class) that captures Position and batchIndex.
client's MessageId and MessageIdAdv are slightly different and contain more state that's not relevant here.

lhotari · 2025-10-13T07:57:31Z

pip/pip-423.md

+    // ... existing methods
+    CompletableFuture<Void> skipMessages(int numMessagesToSkip);
+
+    CompletableFuture<Void> skipMessages(Map<String, String> messageIds);


I'm not actually sure if List<MessageIdAdv> is the best option here. I guess we'd need a new type (class) that captures Position and batchIndex.
client's MessageId and MessageIdAdv are slightly different and contain more state that's not relevant here.

lhotari · 2025-10-13T07:58:15Z

pip/pip-423.md

+```java
+// in pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+@Override
+public CompletableFuture<Void> skipMessages(Map<String, String> messageIds) {


should be List of message ids with (ledgerId, entryId and batchIndex information)

lhotari · 2025-10-13T07:59:59Z

pip/pip-423.md

+    List<Position> positions = new ArrayList<>();
+    for (Map.Entry<String, String> entry : messageIds.entrySet()) {
+        try {
+            long ledgerId = Long.parseLong(entry.getKey());
+            long entryId = Long.parseLong(entry.getValue());
+            Position position = PositionFactory.create(ledgerId, entryId);
+            positions.add(position);
+        } catch (Exception e) {
+            return CompletableFuture.failedFuture(new NotAllowedException("Invalid message ID."));
+        }
+    }
+
+    Map<String, Long> properties = Collections.emptyMap();
+    acknowledgeMessage(positions, AckType.Individual, properties);


this logic should cover handling the case when the batchIndex is passed. If no batchIndex is passed, it should acknowledge the complete entry.

…y MessageId and batch index support

Copilot

Pull Request Overview

This PR introduces PIP-423, which proposes adding a new admin API to skip/acknowledge individual messages by their MessageId. The proposal addresses operational needs where administrators must remove specific problematic messages from subscription backlogs, cancel scheduled delayed messages, or handle poison pill messages without affecting surrounding valid messages.

Key changes:

New REST endpoint /admin/v2/persistent/{tenant}/{namespace}/{topic}/subscription/{subName}/skipByMessageIds with flexible JSON payload formats
New pulsar-admin topics skip-messages CLI command supporting message ID specification via triplets or Base64 encoding
Implementation leverages existing individual acknowledgement mechanism (AckType.Individual) for persistence and reliability

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

pip/pip-423.md

Copilot

Pull Request Overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated 3 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

pip/pip-423.md

…and Subscription interface extension

[improve][pip] PIP-423: Add Support for Cancelling Individual Delayed…

c3a120a

… Messages

github-actions bot added PIP doc-not-needed Your PR changes do not impact docs labels Jun 1, 2025

This was referenced Jun 1, 2025

[improve][pip] PIP-423: Add Support for Cancelling Individual Delayed Messages Denovo1998/pulsar#8

Open

[improve][broker] PIP-423: Add a new admin API to acknowledge a single message #23907

Open

liangyepianzhou self-assigned this Jun 2, 2025

liangyepianzhou added this to the 4.1.0 milestone Jun 2, 2025

liangyepianzhou assigned Denovo1998 and unassigned liangyepianzhou Jun 2, 2025

[improve][pip] PIP-423: Add Support for Cancelling Individual Delayed…

b8afb80

… Messages

Implement the delayed message cancellation function through acknowled…

941c655

…ge message.

Denovo1998 mentioned this pull request Jun 8, 2025

[improve][broker] Make maxBatchDeletedIndexToPersist configurable and document other related configs #24392

Merged

4 tasks

expanding the scope of this PIP is not just to achieve cancellation o…

8fb5e0b

…f delayed messages.

Denovo1998 changed the title ~~[improve][pip] PIP-423: Add Support for Cancelling Individual Delayed Messages~~ [improve][pip] PIP-423: Add a new admin API to acknowledge a single message Jun 10, 2025

docs(pip): enhance pip-423.md with skipMessages API details

fa751bf

codelipenghui reviewed Jul 29, 2025

View reviewed changes

pip/pip-423.md Outdated Show resolved Hide resolved

docs(pip-423): Separate endpoint for skipping by message ID

96962a8

Denovo1998 mentioned this pull request Aug 4, 2025

[Enhancement] Future optimization and defect fixing of the delayed message module #24600

Open

2 tasks

coderzc modified the milestones: 4.1.0, 4.2.0 Sep 1, 2025

Update pip-423.md

920aecf

lhotari requested changes Oct 13, 2025

View reviewed changes

lhotari reviewed Oct 13, 2025

View reviewed changes

docs(pip-423): Enhance pip-423.md with details on skipping messages b…

767b603

…y MessageId and batch index support

Copilot AI review requested due to automatic review settings November 9, 2025 09:58

Copilot AI reviewed Nov 9, 2025

View reviewed changes

pip/pip-423.md Show resolved Hide resolved

pip/pip-423.md Show resolved Hide resolved

pip/pip-423.md Show resolved Hide resolved

pip/pip-423.md Show resolved Hide resolved

pip/pip-423.md Show resolved Hide resolved

Denovo1998 mentioned this pull request Nov 9, 2025

[improve][broker] PIP-423: Add a new admin API to acknowledge a single message Denovo1998/pulsar#17

Open

5 tasks

Denovo1998 requested review from codelipenghui, Copilot and lhotari November 9, 2025 10:04

Copilot AI reviewed Nov 9, 2025

View reviewed changes

pip/pip-423.md Show resolved Hide resolved

pip/pip-423.md Show resolved Hide resolved

pip/pip-423.md Show resolved Hide resolved

docs(pip-423): Update pip-423.md with details on new SkipEntry model …

c661836

…and Subscription interface extension

		* HTTP Body Parameters (JSON): A map where keys are `ledgerId` as a string and values are `entryId` as a string.
		* Example Body: `{"12345": "100", "12346": "200"}`

[improve][pip] PIP-423: Add a new admin API to acknowledge a single message #24370

Are you sure you want to change the base?

[improve][pip] PIP-423: Add a new admin API to acknowledge a single message #24370

Uh oh!

Conversation

Denovo1998 commented Jun 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Does this pull request potentially affect one of the following parts:

Documentation

Matching PR in forked repository

Uh oh!

Denovo1998 commented Jun 3, 2025

Uh oh!

Denovo1998 commented Jun 8, 2025

Uh oh!

Uh oh!

lhotari Oct 13, 2025

Choose a reason for hiding this comment

Uh oh!

lhotari Oct 13, 2025

Choose a reason for hiding this comment

Uh oh!

lhotari Oct 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lhotari Oct 13, 2025

Choose a reason for hiding this comment

Uh oh!

lhotari Oct 13, 2025

Choose a reason for hiding this comment

Uh oh!

lhotari Oct 13, 2025

Choose a reason for hiding this comment

Uh oh!

lhotari Oct 13, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Denovo1998 commented Jun 1, 2025 •

edited

Loading

lhotari Oct 13, 2025 •

edited

Loading