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
