Hi Bhavesh can you explain what limit you're referring to? I'm asking because `message.max.bytes` is applied per message not per batch. is there another limit I should be aware of?
thanks On Wed, Sep 3, 2014 at 2:07 PM, Bhavesh Mistry <mistry.p.bhav...@gmail.com> wrote: > Hi Jun, > > We have similar problem. We have variable length of messages. So when we > have fixed size of Batch sometime the batch exceed the limit set on the > brokers (2MB). > > So can Producer have some extra logic to determine the optimal batch size > by looking at configured message.max.bytes value. > > During the metadata update, Producer will get this value from the Broker > for each topic and Producer will check if current batch size reach this > limit than break batch into smaller chunk such way that It would not exceed > limit (unless single message exceed the limit). Basically try to avoid data > loss as much as possible. > > Please let me know what is your opinion on this... > > Thanks, > > Bhavesh > > > On Wed, Sep 3, 2014 at 6:21 AM, Alexis Midon < > alexis.mi...@airbedandbreakfast.com> wrote: > > > Thanks Jun. > > > > I'll create a jira and try to provide a patch. I think this is pretty > > serious. > > > > On Friday, August 29, 2014, Jun Rao <jun...@gmail.com> wrote: > > > > > The goal of batching is mostly to reduce the # RPC calls to the broker. > > If > > > compression is enabled, a larger batch typically implies better > > compression > > > ratio. > > > > > > The reason that we have to fail the whole batch is that the error code > in > > > the produce response is per partition, instead of per message. > > > > > > Retrying individual messages on MessageSizeTooLarge seems reasonable. > > > > > > Thanks, > > > > > > Jun > > > > > > > > > On Fri, Aug 29, 2014 at 4:28 PM, Alexis Midon < > > > alexis.mi...@airbedandbreakfast.com <javascript:;>> wrote: > > > > > > > Could you explain the goals of batches? I was assuming this was > simply > > a > > > > performance optimization, but this behavior makes me think I'm > missing > > > > something. > > > > is a batch more than a list of *independent* messages? > > > > > > > > Why would you reject the whole batch? One invalid message causes the > > loss > > > > of batch.num.messages-1 messages :( > > > > It seems pretty critical to me. > > > > > > > > If ack=0, the producer will never know about it. > > > > If ack !=0, the producer will retry the whole batch. If the issue was > > > > related to data corruption (etc), retries might work. But in the case > > of > > > > "big message", the batch will always be rejected and the producer > will > > > give > > > > up. > > > > > > > > If the messages are indeed considered independent, I think this is a > > > pretty > > > > serious issue. > > > > > > > > I see 2 possible fix approaches: > > > > - the broker could reject only the invalid messages > > > > - the broker could reject the whole batch (like today) but the > producer > > > (if > > > > ack!=0) could retry messages one at a time on exception like > > > > "MessageSizeTooLarge". > > > > > > > > opinions? > > > > > > > > Alexis > > > > > > > > ``` > > > > [2014-08-29 16:00:35,170] WARN Produce request with correlation id 46 > > > > failed due to [test,1]: kafka.common.MessageSizeTooLargeException > > > > (kafka.producer.async.DefaultEventHandler) > > > > [2014-08-29 16:00:35,284] WARN Produce request with correlation id 51 > > > > failed due to [test,0]: kafka.common.MessageSizeTooLargeException > > > > (kafka.producer.async.DefaultEventHandler) > > > > [2014-08-29 16:00:35,392] WARN Produce request with correlation id 56 > > > > failed due to [test,0]: kafka.common.MessageSizeTooLargeException > > > > (kafka.producer.async.DefaultEventHandler) > > > > [2014-08-29 16:00:35,499] WARN Produce request with correlation id 61 > > > > failed due to [test,1]: kafka.common.MessageSizeTooLargeException > > > > (kafka.producer.async.DefaultEventHandler) > > > > [2014-08-29 16:00:35,603] ERROR Failed to send requests for topics > test > > > > with correlation ids in [43,62] > > > (kafka.producer.async.DefaultEventHandler) > > > > [2014-08-29 16:00:35,603] ERROR Error in handling batch of 3 events > > > > (kafka.producer.async.ProducerSendThread) > > > > kafka.common.FailedToSendMessageException: Failed to send messages > > after > > > 3 > > > > tries. > > > > at > > > > > > > > > > > > > > kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90) > > > > at > > > > > > > > > > > > > > kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:104) > > > > at > > > > > > > > > > > > > > kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:87) > > > > at > > > > > > > > > > > > > > kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:67) > > > > ``` > > > > > > > > > > > > On Thu, Aug 28, 2014 at 7:13 AM, Jun Rao <jun...@gmail.com > > > <javascript:;>> wrote: > > > > > > > > > That's right. If one message in a batch exceeds the size limit, the > > > whole > > > > > batch is rejected. > > > > > > > > > > When determining message.max.bytes, the most important thing to > > > consider > > > > is > > > > > probably memory since currently we need to allocate memory for a > full > > > > > message in the broker and the producer and the consumer client. > > > > > > > > > > Thanks, > > > > > > > > > > Jun > > > > > > > > > > > > > > > On Wed, Aug 27, 2014 at 9:52 PM, Alexis Midon < > > > > > alexis.mi...@airbedandbreakfast.com <javascript:;>> wrote: > > > > > > > > > > > am I miss reading this loop: > > > > > > > > > > > > > > > > > > > > > > > > > > > https://github.com/apache/kafka/blob/0.8.1/core/src/main/scala/kafka/log/Log.scala#L265-L269 > > > > > > > > > > > > it seems like all messages from `validMessages` (which is > > > > > > ByteBufferMessageSet) are NOT appended if one of the message size > > > > exceeds > > > > > > the limit. > > > > > > > > > > > > I hope I'm missing something. > > > > > > > > > > > > > > > > > > > > > > > > On Wed, Aug 27, 2014 at 9:38 PM, Alexis Midon < > > > > > > alexis.mi...@airbedandbreakfast.com <javascript:;>> wrote: > > > > > > > > > > > > > Hi Jun, > > > > > > > > > > > > > > thanks for you answer. > > > > > > > Unfortunately the size won't help much, I'd like to see the > > actual > > > > > > message > > > > > > > data. > > > > > > > > > > > > > > By the way what are the things to consider when deciding on > > > > > > > `message.max.bytes` value? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Wed, Aug 27, 2014 at 9:06 PM, Jun Rao <jun...@gmail.com > > > <javascript:;>> wrote: > > > > > > > > > > > > > >> The message size check is currently only done on the broker. > If > > > you > > > > > > enable > > > > > > >> trace level logging in RequestChannel, you will see the > produce > > > > > request, > > > > > > >> which includes the size of each partition. > > > > > > >> > > > > > > >> Thanks, > > > > > > >> > > > > > > >> Jun > > > > > > >> > > > > > > >> > > > > > > >> On Wed, Aug 27, 2014 at 4:40 PM, Alexis Midon < > > > > > > >> alexis.mi...@airbedandbreakfast.com <javascript:;>> wrote: > > > > > > >> > > > > > > >> > Hello, > > > > > > >> > > > > > > > >> > my brokers are reporting that some received messages exceed > > the > > > > > > >> > `message.max.bytes` value. > > > > > > >> > I'd like to know what producers are at fault but It is > pretty > > > much > > > > > > >> > impossible: > > > > > > >> > - the brokers don't log the content of the rejected messages > > > > > > >> > - the log messages do not contain the IP of the producers > > > > > > >> > - on the consumer side, no exception is thrown (afaik it is > > > > because > > > > > > >> Ack-0 > > > > > > >> > is used). The only kind of notification is to closed the > > > > connection. > > > > > > >> > > > > > > > >> > [1] Do you have any suggestions to track down the guilty > > > producers > > > > > or > > > > > > >> find > > > > > > >> > out the message content? > > > > > > >> > > > > > > > >> > Even though it makes total sense to have the limit defined > and > > > > > applied > > > > > > >> on > > > > > > >> > the brokers, I was thinking that this check could also be > > > applied > > > > by > > > > > > the > > > > > > >> > producers. Some google results suggest that > > `message.max.bytes` > > > > > might > > > > > > be > > > > > > >> > used by the producers but I can't find any trace of that > > > behavior > > > > in > > > > > > the > > > > > > >> > code. > > > > > > >> > > > > > > > >> > The closest thing I have is > > > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://github.com/apache/kafka/blob/0.8.1/core/src/main/scala/kafka/producer/SyncProducer.scala#L67 > > > > > > >> > but it simply logs the message size and content and the log > > > level > > > > is > > > > > > >> trace. > > > > > > >> > > > > > > > >> > [2] could you please confirm if such a producer-side check > > > exists? > > > > > > >> > > > > > > > >> > > > > > > > >> > thanks! > > > > > > >> > > > > > > > >> > Alexis > > > > > > >> > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
