PIP-150: Support read the message of startMessageId position on the broker side
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
Discussion thread: https://lists.apache.org/thread/n3drk2g2oy766qnbtx17knvtssy3tdyl
## Motivation
Currently, the Pulsar-client supports setting the `startMessageId` for Consumer and Reader, and also supports reading the message of `startMessageId` position.
Assume, we have two message id 1,2,3,4 in the topic:
- When we set `earliest` as `startMessageId` value, we can get the message of message id 1
- When we set `latest` as `startMessageId` value, we can't get any message
Sometimes we want to read the message id 4 for the first time, we have only one approach in client:
```
Reader reader = pulsarClient.newReader()
.topic(topicName)
.subscriptionName(subscriptionName)
.startMessageId(MessageId.latest)
.startMessageIdInclusive()
.create();
reader.hasMessageAvailable();
Message msg = reader.readNext(1, TimeUnit.SECONDS);
```
Call `reader.hasMessageAvailable()` before `reader.readNext()` can get the correct message id 4, which include seek action when the `startMessageIdInclusive()` is enabled.
This approach is confusing. If we do this on the broker side, it will make things easier.
## Goal
This PIP proposes support for reading the message of `startMessageId` position on the broker side:
- Add to `Consumer`
- Add to `Reader`
## Implementation
### Protocol
Add a `start_message_id_inclusive` field to `CommandSubscribe` for determine whether to read the message of `startMessageId` position:
```
message CommandSubscribe {
// some fields
// If specified, the subscription will read the message from the start message id position.
optional bool start_message_id_inclusive = 20 [default = false];
}
```
### ManagedCursorImpl
Add a check in `org.apache.bookkeeper.mledger.impl.ManagedCursorImpl#initializeCursorPosition`.
We only need to care that the `startMessageId` is `MessageId.latest` and the`start_message_id_inclusive` is `true`, we get latest position from ledger as `readPosition` value, otherwise if the`start_message_id_inclusive` is `false`, get next position of the latest position as `readPosition` value.
### Client
The `Consumer` and `Reader` support setting the `start_message_id_inclusive` value to `CommandSubscribe` command.
## Compatibility
This feature can have both backward and forward compatibility, this means the users can use any client to request any broker.
Notice that the users still can read the message of the latest position by call `reader.hasMessageAvailable()` before `reader.readNext()`, but this call can be ignored when using the new client and the new broker.
Contributor guide
Assessment
This issue has not been assessed yet.