uuuuu-diwu opened a new issue, #10983:
URL: https://github.com/apache/rocketmq/issues/10983
### Is Your Feature Request Related to a Problem?
We use one LiteTopic as the durable input mailbox of one long-running agent
session. The consumer may have one message currently being executed while more
messages are queued in the same LiteTopic.
Consider the following sequence for one LiteTopic and consumer group:
```text
A: already delivered and currently running in the agent
B: queued
C: queued
--- CancelInput is accepted ---
D: submitted after CancelInput
```
`CancelInput` means:
- stop the currently running work for A after the consumer receives the
cancel command;
- B and C are obsolete and must not enter the application runtime;
- deliver the cancel command durably to the same logical mailbox;
- preserve D and all messages accepted after the cancel boundary.
With normal FIFO delivery, a cancel message appended to the same LiteTopic
is delivered only after B and C, which is too late. The consumer may execute
work that has already been cancelled before it can observe the cancel command.
This is stronger than ordinary message priority. The broker does not need to
forcibly terminate A; it only needs to deliver the cancel command promptly so
that the agent consumer can stop its own active work. The queued messages
before the cancellation boundary may be logically skipped rather than
physically deleted.
The current public gRPC API exposes
`SyncLiteSubscriptionRequest.offset_option`, including `MAX`, but it does not
define an explicit online operation that atomically skips the current backlog
and publishes a durable control message. It is also unclear whether combining
LiteTopic with Priority Message is or will be supported.
### Describe the Solution You'd Like
First, we would like the community to clarify the recommended LiteTopic
pattern for this use case.
If no supported pattern exists, we propose discussing a protocol-level
operation with semantics similar to:
```text
SupersedeQueuedMessagesAndPublishControl(
parent_topic,
lite_topic,
consumer_group,
control_message,
request_id
)
```
The exact API shape is open for discussion. The required behavior is:
1. Establish one cancellation boundary for the specified LiteTopic and
consumer group.
2. Skip ready messages accepted before that boundary, so they are not
delivered to application logic.
3. Persist the cancel control message and make it the next relevant message
delivered to the consumer.
4. Preserve messages accepted after the boundary.
5. Do not claim to revoke a message already delivered to the consumer. The
cancel message tells the consumer to stop that active operation.
6. Make retries idempotent, for example through a request ID.
7. Preserve the result across consumer disconnect, failover, and
resubscription.
An alternative design could be a control/system lane within a LiteTopic,
provided it also defines how earlier queued normal messages are superseded and
how messages submitted after cancellation are preserved.
We are opening a feature issue first to validate the use case and API
boundary. If this requires a new public API or message model, we can help turn
the agreed direction into a RIP.
### Describe Alternatives You've Considered
1. **Append the cancel command to the same LiteTopic**
It remains behind B and C, so obsolete work may execute before
cancellation is observed.
2. **Use a separate control LiteTopic plus an application-side cancellation
fence**
This can work, but introduces a second subscription per session,
cross-channel ordering, durable fence state, and failover coordination. It also
requires draining and acknowledging every superseded input message.
3. **Use Priority Message for the cancel command**
The public API models `LITE` and `PRIORITY` as separate message types and
does not document their composition. Priority alone also does not invalidate
the earlier queued inputs or address an already delivered message.
4. **Synchronize the subscription with `OffsetOption.MAX`, then publish
cancel**
`OffsetOption` is part of `SyncLiteSubscriptionRequest`, not a standalone
online supersede operation. Resetting/skipping and publishing the cancel
command are not expressed atomically, and messages already delivered or
prefetched remain outside that boundary.
5. **Track the active RocketMQ client ID and send it a reset/cancel command**
The public protocol can expose connected client IDs for diagnostics, but
it does not expose a LiteTopic reset/cancel command addressed to a client ID.
Client reconnect and ownership transfer would also make this
application-managed routing state stale.
### Additional Context
This use case comes from long-running AI agents, but it also applies to
session workers, device command queues, workflow executors, and other directed
mailboxes where a later command supersedes queued work.
Related public proposals and protocol:
- RIP-83 Lite Topic:
https://github.com/apache/rocketmq/wiki/RIP%E2%80%9083-Lite-Topic:-A-New-Message-Model
- RIP-80 Priority Message:
https://github.com/apache/rocketmq/wiki/RIP%E2%80%9080-Priority-Message
- LiteTopic gRPC subscription protocol:
https://github.com/apache/rocketmq-apis/blob/master/apache/rocketmq/v2/service.proto
--
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]
