Copilot commented on code in PR #24370:
URL: https://github.com/apache/pulsar/pull/24370#discussion_r2507822180
##########
pip/pip-423.md:
##########
@@ -0,0 +1,278 @@
+# 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.
+* **Batch Messages:** To improve throughput, Pulsar producers can batch
multiple individual messages into a single entry that is written to BookKeeper.
In this case, the `MessageId` also contains a `batchIndex` to identify a
specific message within the batch. The entry's metadata stores the total number
of messages in the batch.
+* **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, or even a single message within a batch. When a message is
acknowledged individually, the broker's `ManagedCursor` persistently tracks
this "acknowledgement hole" 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. 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.
+
+# 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`, `entryId`,
and an optional `batchIndex` for messages within a batch.
+* 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 specifies the topic, target subscription, and a list of message
identifiers. These identifiers can be provided as a triplet of
`ledgerId:entryId:batchIndex` or as Base64-encoded `MessageId` byte arrays.
+
+2. **Broker Receives Request:** The Pulsar broker receives the admin request
for the new endpoint `.../subscription/{subName}/skipByMessageIds`. It parses
the flexible JSON payload and 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(List<SkipEntry> entries)`
on the target `PersistentSubscription` object. For partitioned topics, the
request is scattered to all partition brokers, 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.
+ * For messages specified without a `batchIndex`, it constructs a
`Position` object for the entire entry.
+ * For messages specified with a `batchIndex`, it first reads the entry
from BookKeeper to get the batch metadata (e.g., batch size). It then
constructs a `Position` object that includes an "ack set" (a bitset) indicating
which messages within the batch are being acknowledged.
+ * It calls its internal `acknowledgeMessage()` method with
`AckType.Individual` for all the constructed `Position` objects.
+
+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 implementation introduces a new flexible request DTO, extends the
`Subscription` interface, and implements the core logic in
`PersistentSubscription`.
+
+1. **New Request DTO:** A new class `SkipMessageIdsRequest` is created to
handle polymorphic JSON deserialization on the broker. This allows the API to
accept multiple formats for specifying message IDs.
+
+2. **Subscription Interface Extension:** The `Subscription` interface is
extended with a new method. `SkipEntry` is an internal record holding the
`ledgerId`, `entryId`, and an optional list of `batchIndexes`.
+```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(List<SkipEntry> entries);
+ // ... existing methods
+}
+```
+
+3. **PersistentSubscription Implementation:** The `PersistentSubscription`
class provides the concrete implementation. It differentiates between
full-entry acknowledgements and partial (batch) acknowledgements.
+
+```java
+// in
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+@Override
+public CompletableFuture<Void> skipMessages(List<SkipEntry> entries) {
+ if (Subscription.isCumulativeAckMode(getType())) {
+ return CompletableFuture.failedFuture(new
NotAllowedException("Unsupported subscription type."));
+ }
+
+ // Separate full-entry acks from partial (batchIndex) acks
+ List<Position> fullEntryPositions = new ArrayList<>();
+ Map<String, List<Integer>> partialAckIndexByPos = new HashMap<>();
+ // ... logic to populate these collections from 'entries'
+
+ // If there are partial acks, read the corresponding entries to get batch
metadata
+ if (!partialAckIndexByPos.isEmpty()) {
+ Set<Position> positionsToLoad = ...; // positions for entries with
batch acks
+ cursor.asyncReplayEntries(positionsToLoad, new
AsyncCallbacks.ReadEntriesCallback() {
+ @Override
+ public void readEntriesComplete(List<Entry> readEntries, Object
ctx) {
+ // ... logic for each entry:
+ // 1. Parse MessageMetadata to get batch size.
+ // 2. Validate batch indexes.
+ // 3. Create a BitSet representing the ack state.
+ // 4. Create a Position with the ack set using
AckSetStateUtil.createPositionWithAckSet().
+ // 5. Add this special position to a final list.
+
+ // Finally, acknowledge all positions (full and partial)
+ acknowledgeMessage(finalPositionsList, AckType.Individual,
properties);
Review Comment:
Missing `properties` parameter definition. The code references a
`properties` variable in `acknowledgeMessage(finalPositionsList,
AckType.Individual, properties)` on line 109 and line 116, but this variable is
never defined or initialized in the method. It should be defined or the
parameter should be removed/clarified.
##########
pip/pip-423.md:
##########
@@ -0,0 +1,278 @@
+# 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.
+* **Batch Messages:** To improve throughput, Pulsar producers can batch
multiple individual messages into a single entry that is written to BookKeeper.
In this case, the `MessageId` also contains a `batchIndex` to identify a
specific message within the batch. The entry's metadata stores the total number
of messages in the batch.
+* **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, or even a single message within a batch. When a message is
acknowledged individually, the broker's `ManagedCursor` persistently tracks
this "acknowledgement hole" 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. 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.
+
+# 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`, `entryId`,
and an optional `batchIndex` for messages within a batch.
+* 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 specifies the topic, target subscription, and a list of message
identifiers. These identifiers can be provided as a triplet of
`ledgerId:entryId:batchIndex` or as Base64-encoded `MessageId` byte arrays.
+
+2. **Broker Receives Request:** The Pulsar broker receives the admin request
for the new endpoint `.../subscription/{subName}/skipByMessageIds`. It parses
the flexible JSON payload and 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(List<SkipEntry> entries)`
on the target `PersistentSubscription` object. For partitioned topics, the
request is scattered to all partition brokers, 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.
+ * For messages specified without a `batchIndex`, it constructs a
`Position` object for the entire entry.
+ * For messages specified with a `batchIndex`, it first reads the entry
from BookKeeper to get the batch metadata (e.g., batch size). It then
constructs a `Position` object that includes an "ack set" (a bitset) indicating
which messages within the batch are being acknowledged.
+ * It calls its internal `acknowledgeMessage()` method with
`AckType.Individual` for all the constructed `Position` objects.
+
+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 implementation introduces a new flexible request DTO, extends the
`Subscription` interface, and implements the core logic in
`PersistentSubscription`.
+
+1. **New Request DTO:** A new class `SkipMessageIdsRequest` is created to
handle polymorphic JSON deserialization on the broker. This allows the API to
accept multiple formats for specifying message IDs.
+
+2. **Subscription Interface Extension:** The `Subscription` interface is
extended with a new method. `SkipEntry` is an internal record holding the
`ledgerId`, `entryId`, and an optional list of `batchIndexes`.
+```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(List<SkipEntry> entries);
+ // ... existing methods
+}
+```
+
+3. **PersistentSubscription Implementation:** The `PersistentSubscription`
class provides the concrete implementation. It differentiates between
full-entry acknowledgements and partial (batch) acknowledgements.
+
+```java
+// in
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+@Override
+public CompletableFuture<Void> skipMessages(List<SkipEntry> entries) {
+ if (Subscription.isCumulativeAckMode(getType())) {
+ return CompletableFuture.failedFuture(new
NotAllowedException("Unsupported subscription type."));
+ }
+
+ // Separate full-entry acks from partial (batchIndex) acks
+ List<Position> fullEntryPositions = new ArrayList<>();
+ Map<String, List<Integer>> partialAckIndexByPos = new HashMap<>();
+ // ... logic to populate these collections from 'entries'
+
+ // If there are partial acks, read the corresponding entries to get batch
metadata
+ if (!partialAckIndexByPos.isEmpty()) {
+ Set<Position> positionsToLoad = ...; // positions for entries with
batch acks
+ cursor.asyncReplayEntries(positionsToLoad, new
AsyncCallbacks.ReadEntriesCallback() {
+ @Override
+ public void readEntriesComplete(List<Entry> readEntries, Object
ctx) {
+ // ... logic for each entry:
+ // 1. Parse MessageMetadata to get batch size.
+ // 2. Validate batch indexes.
+ // 3. Create a BitSet representing the ack state.
+ // 4. Create a Position with the ack set using
AckSetStateUtil.createPositionWithAckSet().
+ // 5. Add this special position to a final list.
+
+ // Finally, acknowledge all positions (full and partial)
+ acknowledgeMessage(finalPositionsList, AckType.Individual,
properties);
+ result.complete(null);
+ }
+ // ... handle failures
+ });
+ } else {
+ // Only full-entry acks are present, no need to read entries
+ acknowledgeMessage(fullEntryPositions, AckType.Individual, properties);
Review Comment:
Missing `properties` parameter definition. The code references a
`properties` variable in `acknowledgeMessage(fullEntryPositions,
AckType.Individual, properties)`, but this variable is never defined or
initialized in the method. It should be defined or the parameter should be
removed/clarified.
##########
pip/pip-423.md:
##########
@@ -0,0 +1,278 @@
+# 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.
+* **Batch Messages:** To improve throughput, Pulsar producers can batch
multiple individual messages into a single entry that is written to BookKeeper.
In this case, the `MessageId` also contains a `batchIndex` to identify a
specific message within the batch. The entry's metadata stores the total number
of messages in the batch.
+* **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, or even a single message within a batch. When a message is
acknowledged individually, the broker's `ManagedCursor` persistently tracks
this "acknowledgement hole" 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. 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.
+
+# 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`, `entryId`,
and an optional `batchIndex` for messages within a batch.
+* 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 specifies the topic, target subscription, and a list of message
identifiers. These identifiers can be provided as a triplet of
`ledgerId:entryId:batchIndex` or as Base64-encoded `MessageId` byte arrays.
+
+2. **Broker Receives Request:** The Pulsar broker receives the admin request
for the new endpoint `.../subscription/{subName}/skipByMessageIds`. It parses
the flexible JSON payload and 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(List<SkipEntry> entries)`
on the target `PersistentSubscription` object. For partitioned topics, the
request is scattered to all partition brokers, 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.
+ * For messages specified without a `batchIndex`, it constructs a
`Position` object for the entire entry.
+ * For messages specified with a `batchIndex`, it first reads the entry
from BookKeeper to get the batch metadata (e.g., batch size). It then
constructs a `Position` object that includes an "ack set" (a bitset) indicating
which messages within the batch are being acknowledged.
+ * It calls its internal `acknowledgeMessage()` method with
`AckType.Individual` for all the constructed `Position` objects.
+
+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 implementation introduces a new flexible request DTO, extends the
`Subscription` interface, and implements the core logic in
`PersistentSubscription`.
+
+1. **New Request DTO:** A new class `SkipMessageIdsRequest` is created to
handle polymorphic JSON deserialization on the broker. This allows the API to
accept multiple formats for specifying message IDs.
+
+2. **Subscription Interface Extension:** The `Subscription` interface is
extended with a new method. `SkipEntry` is an internal record holding the
`ledgerId`, `entryId`, and an optional list of `batchIndexes`.
+```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(List<SkipEntry> entries);
+ // ... existing methods
+}
+```
+
+3. **PersistentSubscription Implementation:** The `PersistentSubscription`
class provides the concrete implementation. It differentiates between
full-entry acknowledgements and partial (batch) acknowledgements.
+
+```java
+// in
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+@Override
+public CompletableFuture<Void> skipMessages(List<SkipEntry> entries) {
+ if (Subscription.isCumulativeAckMode(getType())) {
+ return CompletableFuture.failedFuture(new
NotAllowedException("Unsupported subscription type."));
+ }
+
+ // Separate full-entry acks from partial (batchIndex) acks
+ List<Position> fullEntryPositions = new ArrayList<>();
+ Map<String, List<Integer>> partialAckIndexByPos = new HashMap<>();
+ // ... logic to populate these collections from 'entries'
+
+ // If there are partial acks, read the corresponding entries to get batch
metadata
+ if (!partialAckIndexByPos.isEmpty()) {
+ Set<Position> positionsToLoad = ...; // positions for entries with
batch acks
+ cursor.asyncReplayEntries(positionsToLoad, new
AsyncCallbacks.ReadEntriesCallback() {
+ @Override
+ public void readEntriesComplete(List<Entry> readEntries, Object
ctx) {
+ // ... logic for each entry:
+ // 1. Parse MessageMetadata to get batch size.
+ // 2. Validate batch indexes.
+ // 3. Create a BitSet representing the ack state.
+ // 4. Create a Position with the ack set using
AckSetStateUtil.createPositionWithAckSet().
+ // 5. Add this special position to a final list.
+
+ // Finally, acknowledge all positions (full and partial)
+ acknowledgeMessage(finalPositionsList, AckType.Individual,
properties);
+ result.complete(null);
+ }
+ // ... handle failures
+ });
+ } else {
+ // Only full-entry acks are present, no need to read entries
+ acknowledgeMessage(fullEntryPositions, AckType.Individual, properties);
+ return CompletableFuture.completedFuture(null);
+ }
+ return result;
Review Comment:
The `result` CompletableFuture is returned but never defined. The code
should define `CompletableFuture<Void> result = new CompletableFuture<>()`
before the conditional logic to ensure it can be completed in the callback and
returned.
--
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]
