Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Greg, I'm sorry if anything I said sounded like I'm trying to minimize the concerns that's definitely not my intention. My background is in databases, and I share your concerns of muddying transaction semantics (which is already, as you pointed out, is different in Kafka). If calling some

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi, I’m trying to understand the crux of the discussion here. In systems I’m used to, it’s entirely allowed to attempt to add an operation to a transaction, have that operation unambiguously fail, and continue with the transaction safe in the knowledge that the failed operation didn’t make it,

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Thanks Greg, Apologize if we gave the impression that we did not try to address your concerns. It was not my intention to just minimize them. Of course, I just don't see it your way, but that's ok. We don't have to agree. That's why we have a DISCUSS thread to begin with. I don't know the

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Alieh, Yes, I think you understand my intent for the prepare() method. Thanks, Greg On Mon, Jul 22, 2024 at 2:54 AM Alieh Saeedi wrote: > Hi Greg, > > > I appreciate your concerns and comprehensive answer. > > > I am not sure whether I fully understood what you meant or not. You mean, > at

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Greg, I appreciate your concerns and comprehensive answer. I am not sure whether I fully understood what you meant or not. You mean, at the end, the user can go for one of the following scenarios: Either 1) `beginTxn()` and `send(record)` and `commitTxn()` or 2) `beginTxn()` and

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Artem and Matthias, > On the other hand, the effort to prove that > keeping all records in memory won't break some scenarios (and generally > breaking one is enough to cause a lot of pain) seems to be significantly > higher than to prove that setting some flag in some API has pretty much 0 >

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

For catching client side errors this would work IMHO. I am ok with this. We throw before we add the record to the batch. Very clean semantics which should also address the concern of "non-atomic tx"... The exception clearly indicates that the record was not added to the TX, and users can

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey folks, Hopefully not to make this KIP go for another spin :-), but I thought of a modification that might actually address safety concerns over using flags to ignore a vaguely specified class of errors. What if we had the following overload of .send method: void send(ProducerRecord

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Matthias, Thank you for rejecting my suggested alternatives. Your responses are the sorts of things I expected to see summarized in the text of the KIP. I agree with most of your rejections, except this one: > "Estimation" is not sufficient, but we would need to know it exactly. > And that's an

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

I agree with Alieh and Artem -- in the end, why buffer records twice? We effectively want to allow to push some error handling (which I btw consider "business logic") into the producer. IMHO, there is nothing wrong with it. Dropping a poison pill record is no really a violation of atomicity

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Greg, > This makes me think that this IGNORE_SEND_ERRORS covers an arbitrary set of error conditions that may be expanded in the future, possibly to cover the broker side RecordTooLargeException. I don't think it contradicts what I said (the keyword here is "in the future") -- with the

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Artem, Thank you for clarifying as I'm joining the conversation late and may have some misconceptions. > Because of this, a more "complete" solution that > allows ignoring RecordTooLargeException regardless of its origin is > actually incorrect, while a "partial" solution that allows ignoring

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Alieh, Thanks for your response. > what does a user do > after a transaction is failed due to a `too-large-record `exception? They > will submit the same batch without the problematic record again. If they re-submit the same record, they are indicating that this record is an integral part of

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Greg, What you say makes a lot of sense. I just wanted to clarify a couple of subtle points. AL1. There is a functional reason to handle errors that happen on send (oginate in the producer logic in the client) vs. errors that are returned from the broker. The problem is that

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Greg, thanks for the feedback. I can understand your concern about atomicity, but what does a user do after a transaction is failed due to a `too-large-record `exception? They will submit the same batch without the problematic record again. What we are providing is actually a better/more

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, Alieh, thanks for the KIP! And everyone else, thanks for the robust discussion. I understand that there are situations in which users desire that the pipeline "just keep working" and skip errors. However, I question whether it is appropriate to support/encourage this behavior via

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Can we update the KIP to clearly document these decisions? Thanks, Justine On Tue, Jul 9, 2024 at 9:25 AM Andrew Schofield wrote: > Hi Chris, > As it stands, the error handling for transactions in KafkaProducer is not > ideal. There’s no reason why a failed operation should fail a transaction

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Chris, As it stands, the error handling for transactions in KafkaProducer is not ideal. There’s no reason why a failed operation should fail a transaction provided that the application can tell that the operation was not included in the transaction and then make its own decision whether to

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey, just catching up on the latest discussion, and I think we are going in circles? And maybe over-complicate things a little bit too much? My take on the KIP is a follows: - As a minimum, we want to solve client side RecordTooLargeException - If we can tackle other client side errors

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Chris Thanks for the comprehensive answer. I don’t believe that things were better/worse before/after the fix of X/Y. I strongly believe that things get better and better when we intend to improve our current solutions. I am happy that experts, including you, are so cautious and, at the

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Alieh, Can you clarify why broker-side errors shouldn't be covered? The only real rationale I can come up with is that it's easier to implement. "Things were better for Kafka Streams before KAFKA-9279 was fixed" isn't very convincing, because Kafka Streams is not the only user of the Java

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Salut from the KIP’s author Clarifying two points: 1) broker side errors: As far as I remember we are not going to cover the errors originating from the broker! A historical fact: One of the debate points in KIP-1038 was that by defining a producer custom handler, the user may assume that

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Justine, I agree that enumerating a list of errors that should be covered by the KIP is difficult; I was thinking it might be easier if we list the errors that should _not_ be covered by the KIP, and only if we can't define a reasonable heuristic that would cover them without having to

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Chris, I think what you say makes sense. I agree that defining the behavior based on code that can possibly change is not a good idea, and I was trying to get a clearer definition from the KIP's author :) I think it can always be hard to ensure that only specific errors are handled unless

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Alieh, I don't love defining the changes for this KIP in terms of a catch clause in the KafkaProducer class, for two reasons. First, the set of errors that are handled by that clause may shift over time as the code base is modified, and second, it would be fairly opaque to users who want to

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Justine, About the consequences: the consequences will be like when we did not have the fix made in `KAFKA-9279`: silent loss of data! Obviously, when the user intentionally chose to ignore errors, that would not be silent any more. Right? Of course, considering all types of `ApiException`s

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Alieh, If we want to allow any error to be ignored we should probably run through all the errors to make sure they make sense. I just want to feel confident that we aren't just making a decision without considering the consequences carefully. Justine On Tue, Jul 2, 2024 at 12:30 PM Alieh

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Justine, yes we talked about `RecordTooLargeException` as an example, but did we ever limit ourselves to only this specific exception? I think neither in the KIP nor in the PR. As Chris mentioned, this KIP is going to undo what we have done in `KAFKA-9279` in case 1) the user is in a

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey folks, I understand where you are coming from by asking for specific use cases. My understanding based on previous conversations was that there were a few different errors that have been seen. One example I heard some information about was when the record was too large and it fails the batch.

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Chris, thanks for sharing your concerns. 1) About the language of KIP (or maybe later in Javadocs): Is that alright if I write all errors that fall into the `ApiException` category thrown (actually returned) by Producer? 2) About future expansion: do you have any better suggestions for enum

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Alieh and Justine, I'm concerned that we're settling on a definition of "poison pill" that's easiest to tackle right now but may lead to shortcomings down the road. I understand the relationship between this KIP and KAFKA-9279, and I can totally get behind the desire to keep things small,

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Chris and Alieh, My understanding is that this KIP is really only trying to solve an issue of a "poison pill" record that fails send(). We've talked a lot about having a generic framework for all errors, but I don't think that is what this KIP is trying to do. Essentially the request is to undo

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Matthias, Thanks for the valid points you mentioned. I updated the KIP and the PR with: 1) mentioning that the new overloaded `send` throws `IllegalStateException` if the user tries to ignore `send()` errors outside of a transaction. 2) the default implementation in `Producer` interface throws

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Ah, sorry--spoke too soon. The PR doesn't show that errors thrown from Producer::send are handled, but instead, ApiException instances that are caught inside KafkaProducer::doSend and are handled by returning an already-failed future are. I think the same question still applies (is this all we

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Alieh, This KIP has evolved a lot since I last looked at it, but the changes seem well thought-out both in semantics and API. One clarifying question I have is that it looks based on the draft PR that we've narrowed the scope from any error that might take place with producing a record to

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Thanks Alieh, it seems this KIP can just pick between a couple of tradeoffs. Adding an overloaded `send()` as the KIP propose makes sense to me and seems to provides the cleanest solution compare to there options we discussed. Given the explicit name of the passed-in option that highlights

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Justine, Thanks for the suggestion. Making applications to validate every single record is not the best way, from an efficiency point of view. Moreover, between changing the behavior of the Producer in `send` and `commitTnx`, the former seems more reasonable and clean. Bests, Alieh On Thu,

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, I updated the KIP as follows: 1) added `TxnSendOption` with two possible values (NONE, IGNORE_SEND_ERRORS). 2) added the new `send` method with three input parameters to `Producer` and `KafkaProducer`. 3) removed `CommitOption` and `commitTransaction(CommitOption)` from `Producer` and

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Alieh, I see there are two options now. So folks will be discussing the approaches and deciding the best way forward before we vote? I do think there could be a problem with the approach on commit if we get stuck on an earlier error and have more records (potentially on new partitions) to

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Justine, I did not update the KIP with `TxnSendOption` since I thought it'd be better discussed here beforehand. right now, there are 2 PRs: - the PR that implements the current version of the KIP: https://github.com/apache/kafka/pull/16332 - the POC PR that clarifies the `TxnSendOption`:

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Alieh, I think I am a little confused. Are the 3 points above addressed by the KIP or did something change? The PR seems to not include this change and still has the CommitOption as well. Thanks, Justine On Wed, Jun 26, 2024 at 2:15 PM Alieh Saeedi wrote: > Hi all, > > > Looking at the PR

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, Looking at the PR corresponding to the KIP, there are some points worthy of mention: 1) clearing the error ends up dirty/messy code in `TransactionManager`. 2) By clearing the error, we are actually doing an illegal transition from

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, thanks for the brilliant ideas. The KIP is updated as follows: -Regarding the word "latest”: I chose this word because in documentation of the `commitTnx()`, it is clearly mentioned that the method throws the lates exception of `send()`. BTW, I agree with you to change it to “any” since

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi, Looking at your suggestions for the CommitOptions, I would be happy with either. I definitely prefer to the Map in the KIP. We also need to think about the other option where CLEAR_SEND_ERRORS hasn’t been specified, and leave ourselves space for other options in the future. If we use an

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

I was also curious about this text: The new method clears the latest error produced by `send(ProducerRecord)` and transits the transaction back from the error state I agree. We should not say "latest" but "any". Is it fair to say that we expect to encounter only one send response with an

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey there, I had a few questions about the update. Looks like we said on the KIP that the options will be of the type Map commitOptions. Would we want to define something more specific? I was also curious about this text: > The new method clears the latest error produced by

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, Appreciation for maintaining the momentum of our discussion. I see kinda consensus over the main points. It seems that we agreed on the following: 1) Define the `commitTnx(commitOptions)` to clear the error. 2) Make the user explicitly call `flush()` before

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Artem, Yes, I completely agree that by default, special action shouldn't be required from users to prevent transactions from being committed when one or more records can't be sent. The behavior I was suggesting was only relevant to the new API we're discussing where we allow users to

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey folks, Great discussion! Re: throwing exceptions from send(). send() is documented to throw KafkaException, so if the application doesn't handle it, it should be a bug. Now, it does have a note that API exceptions wouldn't be thrown, not sure if we have code that relies on that. There is

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

One quick thought: if a user invokes Producer::abortTransaction from within a producer callback today, even in the midst of an ongoing call to Producer::commitTransaction, what is the behavior? Would it be reasonable to support this behavior as a way to allow error handling to take place during

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

My point it, that it does not seem to be safe to allow users to ignore errors with an implicit flush, and I think it's better to only allow it with (ie, after) an explicit flush(). My reasoning is, that users should make a decision to ignore errors or not, before calling `commitTx()`, but

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Transaction verification is a concept from KIP-890 referring to the verification that a partition has been added to the transaction. It's not a huge deal, but maybe we don't want to overload the terminology. For option 2, I was a little confused by this > when commitTx is called, there is

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Agreed. Options 1 and 3 are safe. Option 2 is not. I’d be happy with 3a as the way. I suggest “TRANSACTION VERIFIED”. There isn’t really precedent for options in the producer API. We could use an enum, which is easy to use and not very future-proof. Or we could use a class like the admin API

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

I am ok either way (ie, flush or commit), but I think we need to define exact semantics, and I think there is some subtle thing to consider: 1) flush(Options) Example: send(...) send(...) flush(ignoreErrors) // at this point, we know that all Futures are completed and all

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Chris, That works for me too. I slightly prefer an option on flush(), but what you suggested works too. Thanks, Andrew > On 24 Jun 2024, at 15:14, Chris Egerton wrote: > > Hi Andrew, > > I like a lot of what you said, but I still believe it's better to override > commitTransaction than

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Andrew, I like a lot of what you said, but I still believe it's better to override commitTransaction than flush. Users will already have to manually opt in to ignoring errors encountered during transactions, and we can document recommended usage (i.e., explicitly invoking flush() before

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Alieh, Thanks for driving this. Unfortunately, there are many parts of the API which are a bit unfortunate and it’s tricky to make small improvements that don’t have downsides. I don’t like the idea of using a configuration because configuration is often outside the application and changing

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, Thanks for the interesting discussion. I assume that now the main questions are as follows: 1. Do we need to transit the transcation to the error state for API exceptions? 2. Should we throw the API exception in `send()` instead of returning a future error? 3. If the answer to question

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hey Kirk, can you elaborate on a few points? Otherwise users would have to know to explicitly change their code to invoke flush(). Why? If we would add an option to `flush(FlushOption)`, the existing `flush()` w/o any option will still be there, right? If we would really deprecate

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Matthias, > On Jun 21, 2024, at 12:28 PM, Matthias J. Sax wrote: > > If we want to limit it to `RecordTooLargeException` throwing from `send()` > directly make sense. Thanks for calling it out. > > It's still a question of backward compatibility? `send()` does throw > exceptions already,

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, The JavaDoc for Producer.flush() states: Applications don't need to call this method for transactional producers, since the commitTransaction() will flush all buffered records before performing the commit. This ensures that all the send() calls made since the previous

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

If we want to limit it to `RecordTooLargeException` throwing from `send()` directly make sense. Thanks for calling it out. It's still a question of backward compatibility? `send()` does throw exceptions already, including generic `KafkaException`. Not sure if this helps with backward

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Artem, I think it'd make sense to throw directly from send whenever possible, instead of returning an already-completed future. I didn't do that in my bug fix to try to be conservative about breaking changes but this seems to have caused its own set of headaches. It would be a little less

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Not sure if we can change send and make it throw, given that send() is async? That is why users can register a `Callback` to begin with, right? And Alieh's point about backward compatibility is also a fair concern. Actually, this would potentially be even worse than the original buggy

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, It is very exciting to see all the experts here raising very good points. As we go further, we see more and more options to improve our solution, which makes concluding and updating the KIP impossible. The main suggestions so far are: 1. `flush` with `flushOptions` as input parameter

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Artem, I think you make a good point which is worth further consideration. If any of the existing methods is really ripe for a change here, it’s the send() that actually caused the problem. If that can be fixed so there are no situations in which a lurking error breaks a transaction, that might

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

> I thought we still wait for requests (and their errors) to come in and could handle fatal errors appropriately. We do wait for requests, but my understanding is that when commitTransaction("ignore send errors") we want to ignore errors. So if we do 1. send 2. commitTransaction("ignore send

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

I'm a bit late to the party, but the discussion here looks reasonable. Moving the logic to a transactional method makes sense to me and makes me feel a bit better about keeping the complexity in the methods relevant to the issue. > One minor concern is that if we set "ignore send errors" (or

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Matthias (and other folks who suggested ideas), > maybe `commitTransaction(CommitOptions)` or similar could be a good way forward? I like this approach. One minor concern is that if we set "ignore send errors" (or whatever we decide to name it) option without explicit flush, it'll actually

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Seems the option to use a config does not get a lot of support. So we need to go with some form or "overload / new method". I think Chris' point about not coupling it to `flush()` but rather `commitTransaction()` is actually a very good one; for non-tx case, the different flush variants would

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Alieh, Thanks for the KIP. I *really* don’t like adding a config which changes the behaviour of the flush() method. We already have too many configs. But I totally understand the problem that you’re trying to solve and some of the other suggestions in this thread seem neater. Personally, I

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, thanks for the KIP Alieh! LM1. Totally agree with Artem's point about the config not being the most explicit/flexible way to express this capability. Getting then to Matthias 4 options, what I don't like about 3 and 4 is that it seems they might not age very well? Aren't we going to be

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Matthias, I like the alternatives you've listed. One more that might help is if, instead of overloading flush(), we overloaded commitTransaction() to something like commitTransaction(boolean tolerateRecordErrors). This seems slightly cleaner in that it takes the behavioral change we want,

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Thanks for the KIP Alieh. I actually like the KIP as-is, but think Arthem raises very good points... Seems we have four options on how to move forward? 1. add config to allow "silent error clearance" as the KIP proposes 2. change flush() to clear error and let it throw 3. add new

Re: [DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi Alieh, Thank you for the KIP. I have a couple of suggestions: AL1. We should throw an error from flush after we clear it. This would make it so that both "send + commit" and "send + flush + commit" (the latter looks like just a more verbose way to express the former, and it would be

[DISCUSS] KIP-1059: Enable the Producer flush() method to clear the latest send() error

Hi all, I'd like to kick off a discussion for KIP-1059 that suggests adding a new feature to the Producer flush() method. https://cwiki.apache.org/confluence/display/KAFKA/KIP-1059%3A+Enable+the+Producer+flush%28%29+method+to+clear+the+latest+send%28%29+error Cheers, Alieh