lhotari commented on code in PR #24370:
URL: https://github.com/apache/pulsar/pull/24370#discussion_r2425475230
##########
pip/pip-423.md:
##########
@@ -0,0 +1,254 @@
+# PIP-423: Add a new admin API to acknowledge a single message
+
+# Background knowledge
+
+* **Message Identification (MessageId):** In Pulsar, every message is
uniquely identified by its `MessageId`, which encapsulates a `ledgerId`, an
`entryId`, and optionally a partition index. The combination of `ledgerId` and
`entryId` (often represented as a `Position`) uniquely identifies a message's
physical storage location within a topic. A producer receives a `MessageId` for
each message it successfully sends.
+* **Subscriptions and Cursors:** Each subscription on a topic maintains its
own "cursor" which tracks consumption progress. For subscription types like
`Shared` or `Key_Shared`, the cursor can track individually acknowledged
messages, even if they are out of order relative to the main consumption
progress marker (`mark-delete position`). The cursor is responsible for
ensuring that acknowledged messages are not redelivered.
+* **Individual Acknowledgement:** Pulsar subscriptions support different
acknowledgement modes. `Shared` and `Key_Shared` subscriptions heavily rely on
**individual acknowledgement**. This allows a consumer to acknowledge a single
message. When a message is acknowledged individually ahead of the `mark-delete
position`, the broker's `ManagedCursor` persistently tracks this to ensure that
acknowledged messages are not redelivered after a broker or consumer restart.
This proposal leverages this existing, robust mechanism.
+* **Delayed Messages:** Pulsar supports scheduling messages for future
delivery. A primary use case for this proposal is to allow a scheduled message
to be effectively "cancelled" by acknowledging it before its delivery time.
Since the message is marked as consumed by the cursor, the
`DelayedDeliveryTracker` will not dispatch it.
+* **Existing `skip` API:** Pulsar has an admin API to `skip` a specified
*number* of messages for a subscription. This is useful for bulk-skipping but
lacks the precision to target a single, specific message within a large
backlog, especially if its exact sequence is unknown. This proposal provides a
more precise way to skip messages by their specific `MessageId`.
+
+# 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.
+* **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`, providing a robust and
reliable way to remove any individual message—delayed or not—from a
subscription's backlog.
+
+# Goals
+
+## In Scope
+
+* Introduce a new Admin API endpoint and a corresponding `pulsar-admin` CLI
command to support skipping specific messages for a subscription.
+* The target message(s) will be identified by their `ledgerId` and `entryId`.
+* The implementation will leverage Pulsar's existing, robust
`AckType.Individual` mechanism for persistence and reliability.
+* This feature will only be supported for subscription types that allow
individual acknowledgements (e.g., `Shared`, `Key_Shared`).
+* Ensure that once a message is successfully skipped via this API, it will
not be delivered to any consumer on the targeted subscription.
+* Support for both partitioned and non-partitioned topics.
+
+## Out of Scope
+
+* Modifying the existing `skip/{numMessages}` endpoint. A new, dedicated
endpoint will be created for clarity.
+* Automatic skipping of messages across geo-replicated clusters. The command
is a per-cluster administrative operation that must be run on each cluster
where the skip is needed.
+
+# High Level Design
+
+The proposed solution introduces a new admin API that triggers Pulsar's
individual acknowledgement capability on demand for specific messages.
+
+1. **Initiate Skip-by-ID:** An administrator initiates the action via the new
`pulsar-admin topics skip-messages` command or its corresponding REST API call.
The request must specify the topic, the target subscription, and a list of
message IDs (`ledgerId:entryId`) to be skipped.
+
+2. **Broker Receives Request:** The Pulsar broker receives the admin request
for the new endpoint `.../subscription/{subName}/skipByMessageIds`. It
validates the administrator's permissions for the topic, re-using the existing
`TopicOperation.SKIP` authorization rule.
+
+3. **Delegate to Subscription:** The broker, after validating topic ownership
and permissions, invokes a new method `skipMessages(Map<String, String>
messageIds)` on the target `PersistentSubscription` object. For partitioned
topics, the request is scattered to all partitions, and each partition broker
performs this action.
+
+4. **Perform Individual Acknowledgement:** Inside the
`PersistentSubscription`, the following occurs:
+ * It verifies that the subscription's type supports individual
acknowledgements.
+ * It constructs `Position` objects from the provided ledger and entry
IDs.
+ * It calls its internal `acknowledgeMessage()` method with
`AckType.Individual` for the specified positions. This is functionally
identical to what would happen if a consumer individually acknowledged the
message.
+
+5. **Persistence and Effect:** The `ManagedCursor` for the subscription
records these individual acknowledgements, which are persisted to metadata
storage.
+ * For a **regular message** in the backlog, it is marked as consumed for
that subscription and will not be delivered.
+ * For a **delayed message**, it is marked as consumed before the
`DelayedDeliveryTracker` attempts to schedule it. The message is thus
effectively **cancelled**.
+
+This design is simple and robust as it builds upon the broker's proven message
acknowledgement foundation while providing a clean, dedicated administrative
API for this precise operational task.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The core of the implementation involves adding a new method to the
`Subscription` interface and implementing it in `PersistentSubscription` to
leverage the existing individual acknowledgment mechanism.
+
+1. **Subscription Interface Extension:**
+ The `Subscription` interface is extended with a new method to handle
skipping by message IDs.
+
+```java
+// in
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java
+public interface Subscription extends MessageExpirer {
+ // ... existing methods
+ CompletableFuture<Void> skipMessages(int numMessagesToSkip);
+
+ CompletableFuture<Void> skipMessages(Map<String, String> messageIds);
+ // ... existing methods
+}
+```
+
+2. **PersistentSubscription Implementation:**
+
+The `PersistentSubscription` class provides the concrete implementation. It
converts the message ID map into `Position` objects and uses the standard
individual acknowledge flow.
+
+```java
+// in
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+@Override
+public CompletableFuture<Void> skipMessages(Map<String, String> messageIds) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] Skipping messages by messageIds, current backlog
{}", topicName, subName,
+ cursor.getNumberOfEntriesInBacklog(false));
+ }
+
+ if (Subscription.isCumulativeAckMode(getType())) {
+ return CompletableFuture.failedFuture(new
NotAllowedException("Unsupported subscription type."));
+ }
+
+ 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);
+
+ return CompletableFuture.completedFuture(null);
+}
+```
+
+3. **Admin API Logic:**
+
+The `PersistentTopicsBase` class is updated with a new
`internalSkipByMessageIds` method which handles the request and calls the
`subscription.skipMessages(Map)` method. It also includes logic to handle
partitioned topics by fanning out the request to each partition owner.
+```java
+// in
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+protected void internalSkipByMessageIds(AsyncResponse asyncResponse, String
subName, boolean authoritative,
+ Map<String, String> messageIds) {
+ // ... validate operation and ownership ...
+
+ // Logic for a single partition / non-partitioned topic
+ // ...
+ Subscription sub = topic.getSubscription(subName);
+ if (sub == null) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ getSubNotFoundErrorMessage(topic.getName(), subName)));
+ }
+ return sub.skipMessages(messageIds);
+ // ...
+}
+```
+
+## Public-facing Changes
+The existing `skipMessages` API is modified to accept a POST body containing
message IDs.
+
+### Public API
+
+A new REST endpoint is added for skipping specific messages.
+
+* **Path (v2):** `POST
/admin/v2/persistent/{tenant}/{namespace}/{topic}/subscription/{subName}/skipByMessageIds`
+* **Path (v1):** `POST
/admin/v1/persistent/{property}/{cluster}/{namespace}/{topic}/subscription/{subName}/skipByMessageIds`
+* **Path Parameters:**
+ * All path parameters identify the target subscription.
+* **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"}`
+* **Response Codes:**
+ * `204 No Content`: The operation was successful.
+ * `400 Bad Request`: Invalid message ID format in the request body
(e.g., non-numeric ledgerId or entryId).
+ * `403 Forbidden`: The client is not authorized to perform a `SKIP`
operation on this topic.
+ * `404 Not Found`: The topic or subscription does not exist.
+ * `405 Method Not Allowed`: The subscription type does not support
individual acknowledgement (e.g., `Exclusive`, `Failover`).
+ * `500 Internal Server Error`: An unexpected error occurred in the
broker.
+
+### Binary protocol
+
+No changes are made to the Pulsar binary protocol.
+
+### Configuration
+
+**No new configuration parameters are introduced. However, this feature's
performance and behavior under heavy load are directly influenced by existing
`ManagedLedger` configurations that govern the persistence of acknowledgement
holes.**
+
+**Administrators should be aware of these settings if they expect a high
volume of message cancellations:**
+
+```
+# Max number of "acknowledgment holes" that are going to be persistently
stored.
+# When acknowledging out of order, a consumer will leave holes that are
supposed
+# to be quickly filled by acking all the messages. The information of which
+# messages are acknowledged is persisted by compressing in "ranges" of messages
+# that were acknowledged. After the max number of ranges is reached, the
information
+# will only be tracked in memory and messages will be redelivered in case of
+# crashes.
+managedLedgerMaxUnackedRangesToPersist=10000
+
+# Maximum number of partially acknowledged batch messages per subscription
that will have their batch
+# deleted indexes persisted. Batch deleted index state is handled when
acknowledgmentAtBatchIndexLevelEnabled=true.
+# When this limit is exceeded, remaining batch message containing the batch
deleted indexes will
+# only be tracked in memory. In case of broker restarts or load balancing
events, the batch
+# deleted indexes will be cleared while redelivering the messages to consumers.
+managedLedgerMaxBatchDeletedIndexToPersist=10000
+
+# When storing acknowledgement state, choose a more compact serialization
format that stores
+# individual acknowledgements as a bitmap which is serialized to an array of
long values. NOTE: This setting requires
+# managedLedgerUnackedRangesOpenCacheSetEnabled=true to be effective.
+managedLedgerPersistIndividualAckAsLongArray=true
+
+# When set to true, a BitSet will be used to track acknowledged messages that
come after the "mark delete position"
+# for each subscription. RoaringBitmap is used as a memory efficient BitSet
implementation for the acknowledged
+# messages tracking. Unacknowledged ranges are the message ranges excluding
the acknowledged messages.
+managedLedgerUnackedRangesOpenCacheSetEnabled=true
+
+# Max number of "acknowledgment holes" that can be stored in MetadataStore. If
number of unack message range is higher
+# than this limit then broker will persist unacked ranges into bookkeeper to
avoid additional data overhead into
+# MetadataStore.
+managedLedgerMaxUnackedRangesToPersistInMetadataStore=1000
+
+# ManagedCursorInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD,
SNAPPY).
+# If value is NONE, then save the ManagedCursorInfo bytes data directly
without compression.
+# Using compression reduces the size of persistent cursor (subscription)
metadata. This enables using a higher
+# managedLedgerMaxUnackedRangesToPersistInMetadataStore value and reduces the
overall metadata stored in
+# the metadata store such as ZooKeeper.
+managedCursorInfoCompressionType=NONE
+
+# ManagedCursorInfo compression size threshold (bytes), only compress metadata
when origin size more then this value.
+# 0 means compression will always apply.
+managedCursorInfoCompressionThresholdInBytes=16384
+
+# ManagedLedgerInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD,
SNAPPY).
+# If value is invalid or NONE, then save the ManagedLedgerInfo bytes data
directly without compression.
+# Using compression reduces the size of the persistent topic metadata. When a
topic contains a large number of
+# individual ledgers in BookKeeper or tiered storage, compression helps
prevent the metadata size from exceeding
+# the maximum size of a metadata store entry (ZNode in ZooKeeper). This also
reduces the overall metadata stored
+# in the metadata store such as ZooKeeper.
+managedLedgerInfoCompressionType=NONE
+
+# ManagedLedgerInfo compression size threshold (bytes), only compress metadata
when origin size more then this value.
+# 0 means compression will always apply.
+managedLedgerInfoCompressionThresholdInBytes=16384
+```
+
+### CLI
+
+A new CLI command is added to `pulsar-admin topics`.
+
+* **Command:** `skip-messages`
+* **Description:** Skip some messages for a subscription by their message
IDs.
+* **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.
Review Comment:
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.
--
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]
