To be clear, I'm merely saying we have not had a scenario for it yet. It's not unreasonable to imagine a scenario where you want the send to have succeeded, perform some action later in the same process call.
*>> There is one more case to consider - what if a single process call sends multiple messages? In this case would we need to call checkpoint after every send() call inside same process() call* Calls to commit are not synchronous and declares an intent to commit. They are evaluated after your process completes and returns. So, having multiple commits is probably redundant. On Thu, Mar 9, 2017 at 1:36 PM, Jagadish Venkatraman <jagadish1...@gmail.com > wrote: > > we use "synchronous" guarantee to ensure that the messages we emit > from samza are delivered to Kafka linearly. > > Samza already guarantees that for you. Currently, two *send()* calls that > produce to the same topic partition are *always* delivered in-order. > (regardless of whether you called commit, batched up or otherwise) > > Do you want ordering multiple send calls across partitions? (We have not > had a scenario for that yet). > > Thanks, > Jagadish > > > > > > On Thu, Mar 9, 2017 at 10:19 AM, Gaurav Agarwal <gauravagarw...@gmail.com> > wrote: > >> There is one more case to consider - what if a single process call sends >> multiple messages? In this case would we need to call checkpoint after >> every send() call inside same process() call.. that seems to be >> problematic, as once checkpointed, there is no safety net against any >> failures in subsequent sends() from same process call. >> >> Thanks for being patient with my questions! >> >> >> ________________________________ >> From: Gaurav Agarwal <gauravagarw...@gmail.com> >> Sent: Thursday, March 9, 2017 11:27:17 PM >> To: dev@samza.apache.org >> Cc: Mukul Gupta; Kshitij Gupta >> Subject: Re: Samza 0.12.0 + synchronous KafkaProducer ? >> >> Hi Jagadish, please find reply inline: >> >> (it appears that there is no easy way today to guarantee ordered delivery >> of messages to Kafka from Samza without consuming the checkpointing >> flexibility). >> >> On Thu, Mar 9, 2017 at 11:01 PM, Jagadish Venkatraman < >> jagadish1...@gmail.com<mailto:jagadish1...@gmail.com>> wrote: >> Hi Gaurav, >> >> >> process->process->....->doWork()->checkpoint->process.. >> >> What does *doWork()* do? Does it actually iterate over accumulated >> in-memory state, and send messages to Kafka? >> >> In process calls, some "work" gets accumulated in memory (think of >> them as strings representing some work). Frequently, many process() calls >> produce identical work - so accumulating them in a set has effect of >> "deduping" those. In window call, we really process the accumulated work >> and call Samza's checkpoint explicitly after that. >> >> *>> I found the configuration 'batch.size' which says that ''a batch size >> of zero will disable batching entirely". We can probably use this property >> to force Kafka client to send every message inline.* >> >> Yes, but I'm not entirely sure it will help your use-case. Just because >> "batch.size" is zero, does not necessarily mean the "send" is synchronous. >> For instance, You can totally have an async-send with a zero batch size. >> It's still possible (and likely) that the message did not make it to the >> broker when the send returns. >> >> Yes you are correct. We would require stronger sync guarantees if we >> need to ensure ordered delivery of message >> >> *>> This exception is only notified back to task on next invocation of >> send(). This is a bit puzzling (a) The exception being thrown is for send >> of a message which did not do anything wrong.* >> >> Thanks for the diligent walkthrough. However, that's perfectly fine as we >> will still preserve Kafka guarantees of atleast once in-order delivery. >> >> *>> what if a checkpoint was called after calling send() of the >> message that caused exception? Will we lose processing of that message? >> Wouldn't it be too late by the time the exception is thrown back to the >> client?* >> >> Great observation! We will never lose processing of that message. The call >> to commit will wait until pending futures or until there's an exception. >> In >> your specific case, the commit call will fail. On a re-start you will >> replay from the previous check-pointed offset. (as of the last successful >> commit) >> >> Thanks for clarifying this. >> >> Users have reported these exact issues when running Samza at scale and >> fixed them. Please refer SAMZA-1069, SAMZA-960. >> >> *>> The point I am driving at is that should the send() method >> in MessageCollector interface provide an optional mechanism to operate it >> in **strictly synchronous mode?* >> >> Even if you did, would n't you still need to tie it with the offset >> commits, changelog commits, store-buffer flushes to disk etc. to achieve >> your guarantees across restarts? task.commit does that. >> >> I think synchronous sends and checkpointing/commits are not entirely >> connected - they are two distinct aspects of the system. In the application >> use case that I've described, we use "synchronous" guarantee to ensure that >> the messages we emit from samza are delivered to Kafka linearly. Whereas >> the checkpoint concept is being used to "batch" up some amount of work and >> then increment the offset once that batch is processed (which I believe is >> the original intent of checkpointing). >> >> If we used up "checkpointing" for guaranteeing ordered delivery, we would >> loose the capability of batching. >> >> Do let us know if you have more follow-ups. >> >> Thanks, >> Jagadish >> >> On Wed, Mar 8, 2017 at 10:16 PM, Gaurav Agarwal <gauravagarw...@gmail.com >> <mailto:gauravagarw...@gmail.com>> >> wrote: >> >> > Hi Jagadish, >> > >> > Thank you for very quick and detailed response. >> > >> > We have already set the task.commit.ms<http://task.commit.ms> = -1 and >> are using the >> > checkpointing >> > mechanism to accumulate some work in memory in order to do it more >> > efficiently in batches. So the flow is >> > process->process->....->doWork()->checkpoint->process.. >> > Doing checkpointing after every process call will defeat the above >> strategy >> > that we have been following in our application. >> > >> > However, looking through Kafka docs, I found the configuration >> 'batch.size' >> > which says that ''a batch size of zero will disable batching entirely". >> We >> > can probably use this property to force Kafka client to send every >> message >> > inline. >> > Does that sound reasonable to you? >> > >> > There is another related question here (and please excuse me if this is >> > stupid one!) - >> > >> > If there is any exception in Samza's KafkaSystemProducer.send() method, >> > that exception is stored in the SourceData object. This exception is >> only >> > notified back to task on next invocation of send(). >> > This is a bit puzzling (a) The exception being thrown is for send of a >> > message which did not do anything wrong - the previous guy broke it! and >> > (b) what if a checkpoint was called after calling send() of the message >> > that caused exception? Will we lose processing of that message? >> Wouldn't it >> > be too late by the time the exception is thrown back to the client? >> > >> > The point I am driving at is that should the send() method in >> > MessageCollector interface provide an optional mechanism to operate it >> in >> > strictly synchronous mode? >> > >> > On Thu, Mar 9, 2017 at 11:25 AM, Jagadish Venkatraman < >> > jagadish1...@gmail.com<mailto:jagadish1...@gmail.com>> wrote: >> > >> > > Gaurav, >> > > >> > > I really appreciate your diligent walkthrough of the code base. Please >> > find >> > > my replies inline. >> > > >> > > *>> I am trying to figure out, how to make our Samza task processing >> > > strictly ordered * >> > > >> > > By default, Samza offers you guaranteed in-order atleast-once >> processing >> > > out-of the box (same semantics as Kafka). To ensure that each send is >> > > "acknowledged" by the broker, you can choose to invoke Samza's >> *commit* >> > at >> > > the end of processing every message. >> > > >> > > *>> We do not want to start processing of next message till it is >> > > guaranteed that our previously emitted messages from samza tasks have >> > been >> > > accepted by Kafka broker. Is there any samza configuration that will >> make >> > > this happen? * >> > > >> > > You can do the following: >> > > A. Set task.commit.ms<http://task.commit.ms> = -1 (This will disable >> auto-commit, and allow you >> > > to >> > > call manual commit). >> > > B. At the end of every *process *or *window* call, you can invoke >> > > *taskCoordinator.commit(RequestScope.CURRENT_TASK);* >> > > >> > > >> > > *>>The `MessageCollector` interface does not expose a 'flush()' method >> > that >> > > we could have called after doing a send() to ensure the delivery of >> > message >> > > to Kafka Broker.* >> > > >> > > This is intentional(to provide an single commit/flush API via the >> > > *taskCoordinator >> > > *abstraction). Invoking *taskCoordinator.commit* will wait on pending >> > > futures, flush buffers, flush state stores and checkpoint offsets. >> > > >> > > Please let us know if we can be of more help! >> > > >> > > Thanks, >> > > Jagadish >> > > >> > > >> > > >> > > >> > > >> > > On Wed, Mar 8, 2017 at 9:12 PM, Gaurav Agarwal < >> gauravagarw...@gmail.com<mailto:gauravagarw...@gmail.com> >> > > >> > > wrote: >> > > >> > > > (correcting recipient address) >> > > > >> > > > On Thu, Mar 9, 2017 at 10:39 AM, Gaurav Agarwal < >> > > gauravagarw...@gmail.com<mailto:gauravagarw...@gmail.com>> >> > > > wrote: >> > > > >> > > > > Hi All, >> > > > > >> > > > > We are trying to upgrade to Kafka 0.12.0. In the process we >> noticed >> > > that >> > > > > the Kafka 0.10.0 KafkaProducer client api does not provide any >> > > > > configuration to send() the messages synchronously. One needs to >> wait >> > > on >> > > > > the returned Future for synchronous guarantees. >> > > > > >> > > > > I am trying to figure out, how to make our Samza task processing >> > > strictly >> > > > > ordered - i.e. we want to process an incoming message and >> optionally >> > > > write >> > > > > back some messages to kafka. We do not want to start processing of >> > next >> > > > > message till it is guaranteed that our previously emitted messages >> > from >> > > > > samza tasks have been accepted by Kafka broker. >> > > > > >> > > > > Is there any samza configuration that will make this happen? The ` >> > > > > MessageCollector` interface does not expose a 'flush()' method >> that >> > we >> > > > > could have called after doing a send() to ensure the delivery of >> > > message >> > > > > to Kafka Broker. (note that `TaskInstanceCollector` - specific >> > > > > implementation of `MessageCollector` interface does provide the >> > > required >> > > > > flush() method) >> > > > > >> > > > > -- >> > > > > cheers, >> > > > > gaurav >> > > > > >> > > > >> > > >> > > >> > > >> > > -- >> > > Jagadish V, >> > > Graduate Student, >> > > Department of Computer Science, >> > > Stanford University >> > > >> > >> >> >> >> -- >> Jagadish V, >> Graduate Student, >> Department of Computer Science, >> Stanford University >> >> > > > -- > Jagadish V, > Graduate Student, > Department of Computer Science, > Stanford University > -- Jagadish V, Graduate Student, Department of Computer Science, Stanford University
