I agree that we must ensure that every pending callback must be completed eventually (timeout or another error is not a problem), because we cannot let the client application hang forever. I believe that the application can perform a flush() explicitly and also wait for every callback to be executed if that is the requirement.
Usually you call close() when: 1. you have a serious problem: you already know that there is a hard error, and you want to close the Producer or the Application and possibly start a new one to recover 2. you are shutting down your application or component: you have control over the callbacks, so you can wait for them to complete So case 2. can be covered by the application. We have to support case 1: fail fast and close (no need for flush()) . In my experience trying to implement "graceful stops" adds only complexity and false hopes to the users. Enrico Il giorno mer 29 set 2021 alle ore 13:58 Nirvana <1572139...@qq.com.invalid> ha scritto: > I agree to try to ensure ”at most once“ when closing。 > > > > That would still get controlled by send timeout, after that, the send > will fail and close should proceed. > This sounds more in line with “at most once”。 > > > ------------------ 原始邮件 ------------------ > 发件人: > "dev" > < > matteo.me...@gmail.com>; > 发送时间: 2021年9月29日(星期三) 下午3:55 > 收件人: "Dev"<dev@pulsar.apache.org>; > > 主题: Re: Correct semantics of producer close > > > > > but equally they might be > > surprised when closeAsync doesn't complete because the pending > messages > > can't be cleared > > That would still get controlled by send timeout, after that, the send > will fail and close should proceed. > > -- > Matteo Merli > <matteo.me...@gmail.com> > > On Wed, Sep 29, 2021 at 12:52 AM Jack Vanlightly > <jvanligh...@splunk.com.invalid> wrote: > > > > I can see both sides of the argument regarding whether to flush > pending > > messages or not. But I think what is definitely in the contract is > not to > > discard any callbacks causing user code to block forever. No matter > what, > > we must always call the callbacks. > > > > Personally, I am in favour of a close operation not flushing pending > > messages (and I define pending here as any message that has a > callback). > > The reason is that if we wait for all pending messages to be sent > then we > > now face a number of edge cases that could cause the close operation > to > > take a very long time to complete. What if the user code really just > needs > > to close the producer right now? If we amend the documentation to > make it > > clear that close does not flush pending messages then the user is now > able > > to explicitly craft the behaviour they need. If they want all messages > > flushed first then chaing flushAsync->closeAsync else just > closeAsync. > > > > Unfortunately I think user expectation, regardless of the current > javadoc, > > is that close would flush everything and in an ideal world it would. > We > > have the Principle of Least Surprise but we also have Safe By Default. > > Users might be surprised that when calling closeAsync, a load of their > > pending messages get ConnectionAlreadyClosed, but equally they might > be > > surprised when closeAsync doesn't complete because the pending > messages > > can't be cleared. Failing pending messages is the safer option. User > code > > must handle failure responses and cannot claim data loss with a > > non-positive response. But if they can't close a producer, that could > > result in a wider impact on their system, not to mention more issues > > created in GitHub. > > > > Jack > > > > On Wed, Sep 29, 2021 at 7:05 AM Joe F <joefranc...@gmail.com> > wrote: > > > > > [ External sender. Exercise caution. ] > > > > > > >I don't think that implementing `closeAsync` with graceful > shutdown > > > logic implies a guarantee of message publishing. Rather, it > guarantees > > > that failures will be the result of a real exception or a > timeout. > > > > > > I think that's beside the point. There > is no definition of "real" > > > exceptions. At that point the app is publishing on a > best effort basis, > > > and there are no guarantees anywhere in client or server. > > > > > > There is no concept of "maybe published". OR > > > "published-if-no_real_errors". What does that even > mean? That is only a > > > can of worms which is going to add to developer confusion and > lead to > > > Pulsar users finding in the worst possible way that something > got lost > > > because it never got published. It's a poor experience > when you find it. > > > I have a real life experience where a user used async APIs (in a > lambda), > > > which hummed along fine. One day much later, the cloud had > a hitch, and > > > they discovered a message was not delivered. > > > > > > I am more concerned about developers discovering at the worst > possible time > > > that ""published-if-no_real_errors" is a concept. > > > > > > My suggestion is to make this simple for developers. > > > > > > ----The sync/async nature of the close() [ or any other API, for > that > > > matter ] is completely orthogonal to the API semantics, > and is just a > > > programmatic choice to deal with how resources are managed > within the > > > program. That's not material here.--- > > > > > > A close() is an action that is shutting down the producer right > now, not > > > even waiting for any acks of inflight messages. A willingness to > lose > > > pending/inflight messages is explicit in that call. The > producer will not > > > be around to deal with errors or to retry failed messages once > close() is > > > invoked. > > > > > > On the contrary, if the client does not want to deal with > message loss, > > > then flush(), stick around to gather the acks, deal with errors > and retries > > > etc and then do close() . Then close() will be just a resource > management > > > action on the client. > > > > > > So update the documentation to reflect that. ---> if close() > is called on a > > > producer with messages pending acks, those messages are left > indoubt. Avoid > > > all mention of flushes, best effort etc. Users must buy > into uncertainty, > > > without any qualifications. > > > > > > I would at all costs avoid using the term "graceful" > anywhere. That word > > > has specific semantics associated with it in the systems/storage > domain , > > > and what is being proposed here is nothing like that. > > > > > > -j > > > > > > > > > On Tue, Sep 28, 2021 at 7:05 PM Yunze Xu > <y...@streamnative.io.invalid> > > > wrote: > > > > > > > It’s a good point that > `ProducerImpl#failPendingBatchMessages` treats > > > > messages in batch container also as pending messages. > > > > > > > > I agree with your definition of "graceful close”. It’s more > like a “at > > > > most once” > > > > semantics, like the original JavaDoc said > > > > > > > > > pending writes will not be retried > > > > > > > > Thanks, > > > > Yunze > > > > > > > > > 2021年9月29日 上午5:24，Michael Marshall < > mikemars...@gmail.com> 写道： > > > > > > > > > > Thanks for bringing this thread to the mailing list, > Yunze. > > > > > > > > > > I think the right change is to update the `closeAsync` > method to first > > > > > flush `batchMessageContainer` and to then > asynchronously wait for the > > > > > `pendingMessages` queue to drain. We could add a new > timeout or rely > > > > > on the already implemented `sendTimeout` config to put > an upper time > > > > > limit on `closeAsync`. My reasoning as well as > responses to Joe and > > > > > Yunze follow: > > > > > > > > > >> we need to define the behavior for how to process > `pendingMessages` > > > > >> and `batchMessageContainer` when producer call > `closeAsync`. > > > > > > > > > > Yes, this is exactly the clarification required, and I > agree that the > > > > > Javadoc is ambiguous and that the implementation > doesn't align with > > > > > the Javadoc. > > > > > > > > > > If we view the Javadoc as binding, then the > fundamental question is > > > > > what messages are "pending". The `pendingMessages` > seem pretty easy to > > > > > classify as "pending" given that they are already in > flight on the > > > > > network. > > > > > > > > > > I also consider `batchMessageContainer` to be > "pending" because a > > > > > client application already has callbacks for the > messages in this > > > > > container. These callbacks are expected to complete > when the batch > > > > > message delivery completes. Since the client > application already has a > > > > > reference to a callback, it isn't a problem that the > producer > > > > > implementation initiates the flush logic. (Note that > the current > > > > > design fails the `pendingMessages` but does not fail > the > > > > > `batchMessageContainer` when `closeAsync` is called, > so the callbacks > > > > > for that container are currently left incomplete > forever if the client > > > > > is closed with an unsent batch. We will need to > address this design in > > > > > the work that comes from this discussion.) > > > > > > > > > > Further, the `ProducerImpl#failPendingMessages` method > includes logic > > > > > to call `ProducerImpl#failPendingBatchMessages`, which > implies that > > > > > these batched, but not sent, messages have been > historically > > > > > considered "pending". > > > > > > > > > > If we view the Javadoc as non-binding, I think my > guiding influence > > > > > for the new design would be that the `closeAsync` > method should result > > > > > in a "graceful" shutdown of the client. > > > > > > > > > >> What exactly does "graceful" convey here? > > > > > > > > > > This is a great question, and will likely drive the > design here. I > > > > > view graceful to mean that the producer attempts to > avoid artificial > > > > > failures. That means trying to drain the queue instead > of > > > > > automatically failing all of the queue's callbacks. > The tradeoff is > > > > > that closing the producer takes longer. This reasoning > would justify > > > > > my claim that we should first flush the > `batchMessageContainer` > > > > > instead of failing the batch without any effort at > delivery, as that > > > > > would be artificial. > > > > > > > > > >> There is no guarantee that either case will ensure > the message > > > > >> is published. > > > > > > > > > > I don't think that implementing `closeAsync` with > graceful shutdown > > > > > logic implies a guarantee of message publishing. > Rather, it guarantees > > > > > that failures will be the result of a real exception > or a timeout. > > > > > Since calling `closeAsync` prevents additional > messages from > > > > > delivering, users leveraging this functionality might > be operating > > > > > with "at most once" delivery semantics where they'd > prefer to deliver > > > > > the messages if possible, but they aren't going to > delay application > > > > > shutdown indefinitely to deliver its last messages. If > users need > > > > > stronger guarantees about whether their messages are > delivered, they > > > > > are probably already using the flush methods to ensure > that the > > > > > producer's queues are empty before calling > `closeAsync`. > > > > > > > > > > I also agree that in all of these cases, we're > assuming that users are > > > > > capturing references to the async callbacks and then > making business > > > > > logic decisions based on the results of those > callbacks. > > > > > > > > > > Thanks, > > > > > Michael > > > > > > > > > > On Tue, Sep 28, 2021 at 4:58 AM Yunze Xu > <y...@streamnative.io.invalid > > > > > > > > wrote: > > > > >> > > > > >> I can’t agree more, just like what I’ve said in PR > 12195: > > > > >> > > > > >>> At any case, when you choose `sendAsync`, you > should always make use > > > > of the returned future to confirm the result of all > messages. In Kafka, > > > > it's the send callback. > > > > >> > > > > >> But I found many users are confused about the > current behavior, > > > > especially > > > > >> those are used to Kafka’s close semantics. They > might expect a simple > > > > try > > > > >> to flush existing messages, which works at a > simple test environment, > > > > even > > > > >> there's no guarantee for exception cases. > > > > >> > > > > >> > > > > >> > > > > >>> 2021年9月28日 下午4:37，Joe F <joefranc...@gmail.com> > 写道： > > > > >>> > > > > >>> Clients should not depend on any of this > behaviour, since the broker > > > > is at > > > > >>> the other end of an unreliable network > connection. The > > > > >>> semantic differences are kind of meaningless > from a usability point, > > > > since > > > > >>> flushing on close =/= published. What > exactly does "graceful" convey > > > > >>> here? Flush the buffer on the > client end and hope it makes it to > > > the > > > > >>> server. > > > > >>> > > > > >>> Is there a difference whether you > flush(or process) pending messages > > > > or > > > > >>> not? There is no guarantee that either case > will ensure the message > > > is > > > > >>> published. > > > > >>> > > > > >>> The only way to ensure that messages are > published is to wait for the > > > > ack. > > > > >>> The correct model should be to wait for return > on the blocking API, > > > or > > > > wait > > > > >>> for future completion of the async API, then > handle any publish > > > errors > > > > and > > > > >>> then only close the producer. > > > > >>> > > > > >>> > > > > >>> On Mon, Sep 27, 2021 at 8:50 PM Yunze Xu > > > <y...@streamnative.io.invalid > > > > > > > > > >>> wrote: > > > > >>> > > > > >>>> Hi all, > > > > >>>> > > > > >>>> Recently I found a PR ( > https://github.com/apache/pulsar/pull/12195 > > <https://github.com/apache/pulsar/pull/12195>>; > < > > > > >>>> > https://github.com/apache/pulsar/pull/12195>) that > > > > >>>> modifies the existing semantics of > producer close. There're already > > > > some > > > > >>>> communications in this PR, but I think > it's better to start a > > > > discussion > > > > >>>> here > > > > >>>> to let more know. > > > > >>>> > > > > >>>> The existing implementation of producer > close is: > > > > >>>> 1. Cancel all timers, including send and > batch container > > > > >>>> (`batchMessageContainer`). > > > > >>>> 2. Complete all pending messages > (`pendingMessages`) with > > > > >>>> `AlreadyCloseException`. > > > > >>>> > > > > >>>> See `ProducerImpl#closeAsync` for details. > > > > >>>> > > > > >>>> But the JavaDoc of `Producer#closeAsync` > is: > > > > >>>> > > > > >>>>> No more writes will be accepted from > this producer. Waits until all > > > > >>>> pending write request are persisted. > > > > >>>> > > > > >>>> Anyway, the document and implementation > are inconsistent. But > > > > specifically, > > > > >>>> we need to define the behavior for how to > process `pendingMessages` > > > > and > > > > >>>> `batchMessageContainer` when producer call > `closeAsync`. > > > > >>>> > > > > >>>> 1. batchMessageContainer: contains the > buffered single messages > > > > >>>> (`Message<T>`). > > > > >>>> 2. pendingMessages: all inflight messages > (`OpSendMsg`) in network. > > > > >>>> > > > > >>>> IMO, from the JavaDoc, only > `pendingMessages` should be processed > > > and > > > > the > > > > >>>> messages in `batchMessageContainer` should > be discarded. > > > > >>>> > > > > >>>> Since other clients might have already > implemented the similar > > > > semantics of > > > > >>>> Java clients. If we changed the semantics > now, the behaviors among > > > > >>>> different > > > > >>>> clients might be inconsistent. > > > > >>>> > > > > >>>> Should we add a configuration to support > graceful close to follow > > > the > > > > >>>> docs? Or > > > > >>>> just change the current behavior? > > > > >>>> > > > > >>>> Thanks, > > > > >>>> Yunze > > > > >> > > > > > > > > > > >
