Thanks Enrico, Good suggestion.
> do you mean that you want to start reading from the latest message? Yes. > I suggest to add more messages to the example, in order to make it clearer > (like 1,2,3,4,5) Added. > I am not sure if this will change the semantics of existing programs. Your worry is necessary. I need to know more people's views. > Shall we have to add a new configuration option? I think we should add a new configuration here. Do you mean that you start reading from the latest message when using the `latest`? If so, the old behavior will be broken. Enrico Olivelli <eolive...@gmail.com> 于2022年3月26日周六 05:42写道： > Il Ven 25 Mar 2022, 17:36 Zixuan Liu <node...@gmail.com> ha scritto: > > > Hi Pulsar community, > > > > I create a PIP for support read the message of `startMessageId` position > on > > the broker side. > > > > The proposal can be found: https://github.com/apache/pulsar/issues/14883 > > > > --------- > > ## 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 and 2: > > > > - When we set `earliest` as `strartMessageId` value, we can the message > of > > message id 1 > > - When we set `latest` as as `strartMessageId` value, we cannot get any > > message > > > > Sometimes we want to read the message id 2 for the first time, > > do you mean that you want to start reading from the latest message? > > I suggest to add more messages to the example, in order to make it clearer > (like 1,2,3,4,5) > > > > we have only > > one approach in client: > > > > ``` > > Reader<byte[]> reader = pulsarClient.newReader() > > .topic(topicName) > > .subscriptionName(subscriptionName) > > .startMessageId(MessageId.latest) > > .startMessageIdInclusive() > > .create(); > > > > reader.hasMessageAvailable(); > > Message<byte[]> msg = reader.readNext(1, TimeUnit.SECONDS); > > ``` > > > > Call `reader.hasMessageAvailable()` before `reader.readNext()`, which can > > seek to the message id 2, this approach seems confusing. > > > Agreed > > > If we do this on > > the broker side, which will make things easier. > > > > ## Goal > > > > This PIP proposes support read the message of `startMessageId` position > on > > the broker side. > > > > ## 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. > > > > I am not sure if this will change the semantics of existing programs. > > Shall we have to add a new configuration option? > > Otherwise programs written the way you are describing won't work anymore > > Enrico > > > > > > --------- > > > > Thanks, > > Zixuan > > >
