You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@pulsar.apache.org by Yunze Xu <yz...@streamnative.io.INVALID> on 2021/10/01 03:26:22 UTC
Re: Correct semantics of producer close
You're right that before a CommandCloseProducer request was completed, the
pending messages should be persisted before this close request was completed
in normal cases. It’s guaranteed by broker side.
Then there’s no inconsistency between the implementation and JavaDocs now.
The key point is whether should we flush the messages in batch container. I prefer
keeping the current semantics. But I found the messages in batch container
never failed. We need to fix the problem. For example, here’s my unit test:
```java
@Test
public void test() throws Exception {
final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
.topic("topic")
.batchingMaxMessages(10000)
.batchingMaxBytes(10000000)
.batchingMaxPublishDelay(100, TimeUnit.SECONDS)
.sendTimeout(1, TimeUnit.SECONDS)
.create();
final CountDownLatch latch = new CountDownLatch(10);
final Map<Integer, Throwable> throwableMap = new ConcurrentHashMap<>();
for (int i = 0; i < 10; i++) {
final Integer index = i;
producer.sendAsync("msg-" + i).whenComplete((id, e) -> {
if (e != null) {
throwableMap.put(index, e);
}
latch.countDown();
});
}
producer.close();
latch.await();
throwableMap.forEach((i, e) -> {
log.info("Message {} failed with {}", i, e);
});
}
```
The test would block forever.
> 2021年10月1日 上午4:22，Michael Marshall <mi...@gmail.com> 写道：
>
> Following up here. I am pretty sure part of this conversation has been
> based on a misunderstanding of the code. From what I can tell, the
> behavior for `Producer#closeAsync` in the client (mostly) aligns with
> the current Javadocs.
>
>> 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`.
>
> I agree with 1, but I think 2 is only partially correct. The client
> will only exceptionally complete pending messages if the connection is
> null or not ready, or after the broker responds to the
> `CLOSE_PRODUCER` command or a timeout passes. This behavior seems
> right to me.
>
> Here is the relevant code:
> https://github.com/apache/pulsar/blob/9d309145f342bc416b8b4663125e1216903a3e83/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L875-L909
>
> The only remaining question: does close imply flush? If not, we'll
> need update the logic to fail the messages contained in the
> `batchMessageContainer` during close. Otherwise, we'll update the
> logic to call flush before sending the `CLOSE_PRODUCER` command and
> everything should work as documented. In both cases, we should update
> the Javadocs to make the behavior clearer.
>
> Thanks,
> Michael
>
>
>
> On Thu, Sep 30, 2021 at 11:55 AM Michael Marshall <mi...@gmail.com> wrote:
>>
>> I have two questions:
>>
>> 1. Does close imply immediate shutdown?
>> 2. Does close imply flush?
>>
>> There is not yet consensus on 1, and 2 is only relevant if 1's answer is "no".
>>
>> Thus far, the conversation has centered on the `Producer#close`
>> method. I'd like to broaden the discussion to include some other
>> methods from the `PulsarClient` interface: `shutdown` and `close`.
>>
>> The Javadoc for `PulsarClient#shutdown` describes the "shutdown
>> immediately" behavior. It says:
>>
>>> Release all the resources and close all the producer, consumer and
>>> reader instances without waiting for ongoing operations to complete.
>>
>> The Javadoc for `PulsarClient#close` describes waiting for
>> pending/in-flight messages to complete before returning. It says:
>>
>>> This operation will trigger a graceful close of all producer, consumer
>>> and reader instances that this client has currently active. That implies
>>> that close will block and wait until all pending producer send requests
>>> are persisted.
>>
>> One question that follows from the above: why does the `Producer` not
>> have a `shutdown` method? I think this is because the "immediate
>> shutdown" behavior is not necessary for a single producer. When
>> immediate shutdown semantics are required, the `PulsarClient#shutdown`
>> method is sufficient because it is used when shutting down the whole
>> application. (If this is not correct, perhaps we should add a
>> `shutdown` method to the producer?)
>>
>> Since immediate shutdown semantics are already available via our
>> client API, I posit that the answer to question 1 is no, `close` does
>> not imply immediate shutdown. At the very least, `close` in the Pulsar
>> Client has not historically implied immediate shutdown.
>>
>> Additionally, it is relevant to point out that the `Producer#close`
>> method is already sending a `CLOSE_PRODUCER` command and waiting on a
>> response back from the broker. The broker's producer close method has
>> the following Javadoc:
>>
>>> Close the producer immediately if: a. the connection is dropped
>>> b. it's a graceful close and no pending publish acks are left
>>> else wait for pending publish acks
>>
>> Since we're already waiting on the broker to respond to the producer's
>> `CLOSE_PRODUCER` request, I see no reason to fail pending/in-flight
>> messages immediately, especially because we should get a response back
>> for those messages before getting the `SUCCESS` response from the
>> broker since the responses will come on the same TCP connection. We
>> could even simplify the close logic so that when the `CLOSE_PRODUCER`
>> request completes (either successfully or because of a failure), we
>> fail all remaining pending message futures.
>>
>> Ultimately, we need to decide whether to update the implementation to
>> match the existing Javadocs, or to update the Javadocs to indicate
>> that `close` means an immediate shutdown, which includes failing all
>> outstanding message futures immediately. My vote is to make the
>> implementation align with the Javadocs.
>>
>> Regarding question 2, I prefer that `close` implies flush because it
>> is only a single (batched) message being flushed. If we do flush this
>> message, we'll need to make sure that the message is sent before
>> the `CLOSE_PRODUCER` command is sent.
>>
>> Thanks,
>> Michael
>>
>>
>> On Wed, Sep 29, 2021 at 7:04 AM Enrico Olivelli <eo...@gmail.com> wrote:
>>>
>>> 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 <15...@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.merli@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.merli@gmail.com>
>>>>
>>>> On Wed, Sep 29, 2021 at 12:52 AM Jack Vanlightly
>>>> <jvanlightly@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 <joefrancisk@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
>>>> <yzxu@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 <
>>>> mikemarsh17@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
>>>> <yzxu@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 <joefrancisk@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
>>>> > > <yzxu@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
>>>> > > > >>
>>>> > > >
>>>> > > >
>>>> > >
Re: Correct semantics of producer close
Posted by Joe F <jo...@gmail.com>.
I am with Enrrico on this, and for more or less the same reasons.
My vote is to update the doc to say that on close(), there are no
guarantees on pending messages.
There is no reasonable basis for any user to claim that messages are not
getting flushed, because there are no guarantees that the flush will
result in successful publishing. User expectations and reality are
different, even with today's implementation.
if someone really requires a flush within close(), that's already
possible by doing a flush before the close.
The best thing to do is to update the close() documentation. (1) Remove
mention of flushing from the javadoc, and (2) add that close does not
guarantee that outstanding messages are published successfully.
On Thu, Sep 30, 2021 at 9:52 PM Michael Marshall <mi...@gmail.com>
wrote:
> Thanks for your analysis, Yunze. I identified above that the messages
> in the batch container were not getting completed correctly, so I put
> together a PR to fix the problematic behavior. This PR will be valid
> regardless of our decision to add flush logic to the close method.
>
> Here is the PR: https://github.com/apache/pulsar/pull/12259.
>
> Thanks,
> Michael
>
>
> On Thu, Sep 30, 2021 at 10:27 PM Yunze Xu <yz...@streamnative.io.invalid>
> wrote:
> >
> > You're right that before a CommandCloseProducer request was completed,
> the
> > pending messages should be persisted before this close request was
> completed
> > in normal cases. It’s guaranteed by broker side.
> >
> > Then there’s no inconsistency between the implementation and JavaDocs
> now.
> >
> > The key point is whether should we flush the messages in batch
> container. I prefer
> > keeping the current semantics. But I found the messages in batch
> container
> > never failed. We need to fix the problem. For example, here’s my unit
> test:
> >
> > ```java
> > @Test
> > public void test() throws Exception {
> > final Producer<String> producer =
> pulsarClient.newProducer(Schema.STRING)
> > .topic("topic")
> > .batchingMaxMessages(10000)
> > .batchingMaxBytes(10000000)
> > .batchingMaxPublishDelay(100, TimeUnit.SECONDS)
> > .sendTimeout(1, TimeUnit.SECONDS)
> > .create();
> > final CountDownLatch latch = new CountDownLatch(10);
> > final Map<Integer, Throwable> throwableMap = new
> ConcurrentHashMap<>();
> > for (int i = 0; i < 10; i++) {
> > final Integer index = i;
> > producer.sendAsync("msg-" + i).whenComplete((id, e) -> {
> > if (e != null) {
> > throwableMap.put(index, e);
> > }
> > latch.countDown();
> > });
> > }
> > producer.close();
> > latch.await();
> > throwableMap.forEach((i, e) -> {
> > log.info("Message {} failed with {}", i, e);
> > });
> > }
> > ```
> >
> > The test would block forever.
> >
> > > 2021年10月1日 上午4:22，Michael Marshall <mi...@gmail.com> 写道：
> > >
> > > Following up here. I am pretty sure part of this conversation has been
> > > based on a misunderstanding of the code. From what I can tell, the
> > > behavior for `Producer#closeAsync` in the client (mostly) aligns with
> > > the current Javadocs.
> > >
> > >> 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`.
> > >
> > > I agree with 1, but I think 2 is only partially correct. The client
> > > will only exceptionally complete pending messages if the connection is
> > > null or not ready, or after the broker responds to the
> > > `CLOSE_PRODUCER` command or a timeout passes. This behavior seems
> > > right to me.
> > >
> > > Here is the relevant code:
> > >
> https://github.com/apache/pulsar/blob/9d309145f342bc416b8b4663125e1216903a3e83/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L875-L909
> > >
> > > The only remaining question: does close imply flush? If not, we'll
> > > need update the logic to fail the messages contained in the
> > > `batchMessageContainer` during close. Otherwise, we'll update the
> > > logic to call flush before sending the `CLOSE_PRODUCER` command and
> > > everything should work as documented. In both cases, we should update
> > > the Javadocs to make the behavior clearer.
> > >
> > > Thanks,
> > > Michael
> > >
> > >
> > >
> > > On Thu, Sep 30, 2021 at 11:55 AM Michael Marshall <
> mikemarsh17@gmail.com> wrote:
> > >>
> > >> I have two questions:
> > >>
> > >> 1. Does close imply immediate shutdown?
> > >> 2. Does close imply flush?
> > >>
> > >> There is not yet consensus on 1, and 2 is only relevant if 1's answer
> is "no".
> > >>
> > >> Thus far, the conversation has centered on the `Producer#close`
> > >> method. I'd like to broaden the discussion to include some other
> > >> methods from the `PulsarClient` interface: `shutdown` and `close`.
> > >>
> > >> The Javadoc for `PulsarClient#shutdown` describes the "shutdown
> > >> immediately" behavior. It says:
> > >>
> > >>> Release all the resources and close all the producer, consumer and
> > >>> reader instances without waiting for ongoing operations to complete.
> > >>
> > >> The Javadoc for `PulsarClient#close` describes waiting for
> > >> pending/in-flight messages to complete before returning. It says:
> > >>
> > >>> This operation will trigger a graceful close of all producer,
> consumer
> > >>> and reader instances that this client has currently active. That
> implies
> > >>> that close will block and wait until all pending producer send
> requests
> > >>> are persisted.
> > >>
> > >> One question that follows from the above: why does the `Producer` not
> > >> have a `shutdown` method? I think this is because the "immediate
> > >> shutdown" behavior is not necessary for a single producer. When
> > >> immediate shutdown semantics are required, the `PulsarClient#shutdown`
> > >> method is sufficient because it is used when shutting down the whole
> > >> application. (If this is not correct, perhaps we should add a
> > >> `shutdown` method to the producer?)
> > >>
> > >> Since immediate shutdown semantics are already available via our
> > >> client API, I posit that the answer to question 1 is no, `close` does
> > >> not imply immediate shutdown. At the very least, `close` in the Pulsar
> > >> Client has not historically implied immediate shutdown.
> > >>
> > >> Additionally, it is relevant to point out that the `Producer#close`
> > >> method is already sending a `CLOSE_PRODUCER` command and waiting on a
> > >> response back from the broker. The broker's producer close method has
> > >> the following Javadoc:
> > >>
> > >>> Close the producer immediately if: a. the connection is dropped
> > >>> b. it's a graceful close and no pending publish acks are left
> > >>> else wait for pending publish acks
> > >>
> > >> Since we're already waiting on the broker to respond to the producer's
> > >> `CLOSE_PRODUCER` request, I see no reason to fail pending/in-flight
> > >> messages immediately, especially because we should get a response back
> > >> for those messages before getting the `SUCCESS` response from the
> > >> broker since the responses will come on the same TCP connection. We
> > >> could even simplify the close logic so that when the `CLOSE_PRODUCER`
> > >> request completes (either successfully or because of a failure), we
> > >> fail all remaining pending message futures.
> > >>
> > >> Ultimately, we need to decide whether to update the implementation to
> > >> match the existing Javadocs, or to update the Javadocs to indicate
> > >> that `close` means an immediate shutdown, which includes failing all
> > >> outstanding message futures immediately. My vote is to make the
> > >> implementation align with the Javadocs.
> > >>
> > >> Regarding question 2, I prefer that `close` implies flush because it
> > >> is only a single (batched) message being flushed. If we do flush this
> > >> message, we'll need to make sure that the message is sent before
> > >> the `CLOSE_PRODUCER` command is sent.
> > >>
> > >> Thanks,
> > >> Michael
> > >>
> > >>
> > >> On Wed, Sep 29, 2021 at 7:04 AM Enrico Olivelli <eo...@gmail.com>
> wrote:
> > >>>
> > >>> 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
> <15...@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.merli@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.merli@gmail.com>
> > >>>>
> > >>>> On Wed, Sep 29, 2021 at 12:52 AM Jack Vanlightly
> > >>>> <jvanlightly@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 <joefrancisk@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
> > >>>> <yzxu@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 <
> > >>>> mikemarsh17@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
> > >>>> <yzxu@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 <
> joefrancisk@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
> > >>>> > > <yzxu@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
> > >>>> > > > >>
> > >>>> > > >
> > >>>> > > >
> > >>>> > >
> >
>
Re: Correct semantics of producer close
Posted by Michael Marshall <mi...@gmail.com>.
Thanks for your analysis, Yunze. I identified above that the messages
in the batch container were not getting completed correctly, so I put
together a PR to fix the problematic behavior. This PR will be valid
regardless of our decision to add flush logic to the close method.
Here is the PR: https://github.com/apache/pulsar/pull/12259.
Thanks,
Michael
On Thu, Sep 30, 2021 at 10:27 PM Yunze Xu <yz...@streamnative.io.invalid> wrote:
>
> You're right that before a CommandCloseProducer request was completed, the
> pending messages should be persisted before this close request was completed
> in normal cases. It’s guaranteed by broker side.
>
> Then there’s no inconsistency between the implementation and JavaDocs now.
>
> The key point is whether should we flush the messages in batch container. I prefer
> keeping the current semantics. But I found the messages in batch container
> never failed. We need to fix the problem. For example, here’s my unit test:
>
> ```java
> @Test
> public void test() throws Exception {
> final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
> .topic("topic")
> .batchingMaxMessages(10000)
> .batchingMaxBytes(10000000)
> .batchingMaxPublishDelay(100, TimeUnit.SECONDS)
> .sendTimeout(1, TimeUnit.SECONDS)
> .create();
> final CountDownLatch latch = new CountDownLatch(10);
> final Map<Integer, Throwable> throwableMap = new ConcurrentHashMap<>();
> for (int i = 0; i < 10; i++) {
> final Integer index = i;
> producer.sendAsync("msg-" + i).whenComplete((id, e) -> {
> if (e != null) {
> throwableMap.put(index, e);
> }
> latch.countDown();
> });
> }
> producer.close();
> latch.await();
> throwableMap.forEach((i, e) -> {
> log.info("Message {} failed with {}", i, e);
> });
> }
> ```
>
> The test would block forever.
>
> > 2021年10月1日 上午4:22，Michael Marshall <mi...@gmail.com> 写道：
> >
> > Following up here. I am pretty sure part of this conversation has been
> > based on a misunderstanding of the code. From what I can tell, the
> > behavior for `Producer#closeAsync` in the client (mostly) aligns with
> > the current Javadocs.
> >
> >> 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`.
> >
> > I agree with 1, but I think 2 is only partially correct. The client
> > will only exceptionally complete pending messages if the connection is
> > null or not ready, or after the broker responds to the
> > `CLOSE_PRODUCER` command or a timeout passes. This behavior seems
> > right to me.
> >
> > Here is the relevant code:
> > https://github.com/apache/pulsar/blob/9d309145f342bc416b8b4663125e1216903a3e83/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L875-L909
> >
> > The only remaining question: does close imply flush? If not, we'll
> > need update the logic to fail the messages contained in the
> > `batchMessageContainer` during close. Otherwise, we'll update the
> > logic to call flush before sending the `CLOSE_PRODUCER` command and
> > everything should work as documented. In both cases, we should update
> > the Javadocs to make the behavior clearer.
> >
> > Thanks,
> > Michael
> >
> >
> >
> > On Thu, Sep 30, 2021 at 11:55 AM Michael Marshall <mi...@gmail.com> wrote:
> >>
> >> I have two questions:
> >>
> >> 1. Does close imply immediate shutdown?
> >> 2. Does close imply flush?
> >>
> >> There is not yet consensus on 1, and 2 is only relevant if 1's answer is "no".
> >>
> >> Thus far, the conversation has centered on the `Producer#close`
> >> method. I'd like to broaden the discussion to include some other
> >> methods from the `PulsarClient` interface: `shutdown` and `close`.
> >>
> >> The Javadoc for `PulsarClient#shutdown` describes the "shutdown
> >> immediately" behavior. It says:
> >>
> >>> Release all the resources and close all the producer, consumer and
> >>> reader instances without waiting for ongoing operations to complete.
> >>
> >> The Javadoc for `PulsarClient#close` describes waiting for
> >> pending/in-flight messages to complete before returning. It says:
> >>
> >>> This operation will trigger a graceful close of all producer, consumer
> >>> and reader instances that this client has currently active. That implies
> >>> that close will block and wait until all pending producer send requests
> >>> are persisted.
> >>
> >> One question that follows from the above: why does the `Producer` not
> >> have a `shutdown` method? I think this is because the "immediate
> >> shutdown" behavior is not necessary for a single producer. When
> >> immediate shutdown semantics are required, the `PulsarClient#shutdown`
> >> method is sufficient because it is used when shutting down the whole
> >> application. (If this is not correct, perhaps we should add a
> >> `shutdown` method to the producer?)
> >>
> >> Since immediate shutdown semantics are already available via our
> >> client API, I posit that the answer to question 1 is no, `close` does
> >> not imply immediate shutdown. At the very least, `close` in the Pulsar
> >> Client has not historically implied immediate shutdown.
> >>
> >> Additionally, it is relevant to point out that the `Producer#close`
> >> method is already sending a `CLOSE_PRODUCER` command and waiting on a
> >> response back from the broker. The broker's producer close method has
> >> the following Javadoc:
> >>
> >>> Close the producer immediately if: a. the connection is dropped
> >>> b. it's a graceful close and no pending publish acks are left
> >>> else wait for pending publish acks
> >>
> >> Since we're already waiting on the broker to respond to the producer's
> >> `CLOSE_PRODUCER` request, I see no reason to fail pending/in-flight
> >> messages immediately, especially because we should get a response back
> >> for those messages before getting the `SUCCESS` response from the
> >> broker since the responses will come on the same TCP connection. We
> >> could even simplify the close logic so that when the `CLOSE_PRODUCER`
> >> request completes (either successfully or because of a failure), we
> >> fail all remaining pending message futures.
> >>
> >> Ultimately, we need to decide whether to update the implementation to
> >> match the existing Javadocs, or to update the Javadocs to indicate
> >> that `close` means an immediate shutdown, which includes failing all
> >> outstanding message futures immediately. My vote is to make the
> >> implementation align with the Javadocs.
> >>
> >> Regarding question 2, I prefer that `close` implies flush because it
> >> is only a single (batched) message being flushed. If we do flush this
> >> message, we'll need to make sure that the message is sent before
> >> the `CLOSE_PRODUCER` command is sent.
> >>
> >> Thanks,
> >> Michael
> >>
> >>
> >> On Wed, Sep 29, 2021 at 7:04 AM Enrico Olivelli <eo...@gmail.com> wrote:
> >>>
> >>> 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 <15...@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.merli@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.merli@gmail.com>
> >>>>
> >>>> On Wed, Sep 29, 2021 at 12:52 AM Jack Vanlightly
> >>>> <jvanlightly@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 <joefrancisk@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
> >>>> <yzxu@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 <
> >>>> mikemarsh17@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
> >>>> <yzxu@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 <joefrancisk@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
> >>>> > > <yzxu@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
> >>>> > > > >>
> >>>> > > >
> >>>> > > >
> >>>> > >
>