Actually, with the new producer, you can configure the batch size in bytes. If you set the batch size to be smaller than the max message size, messages exceeding the max message limit will be in its own batch. Then, only one message will be rejected by the broker.
Thanks, Jun On Tue, Sep 9, 2014 at 5:51 PM, Bhavesh Mistry <mistry.p.bhav...@gmail.com> wrote: > Hi Jun, > > Is there any plug-ability that Developer can customize batching logic or > inject custom code for this ? Shall I file Jira for this issues. > > Thanks, > > Bhavesh > > On Tue, Sep 9, 2014 at 3:52 PM, Jun Rao <jun...@gmail.com> wrote: > > > No, the new producer doesn't address that problem. > > > > Thanks, > > > > Jun > > > > On Tue, Sep 9, 2014 at 12:59 PM, Bhavesh Mistry < > > mistry.p.bhav...@gmail.com> > > wrote: > > > > > HI Jun, > > > > > > Thanks for clarification. Follow up questions, does new producer solve > > the > > > issues highlight. In event of compression and async mode in new > > producer, > > > will it break down messages to this UPPER limit and submit or new > > producer > > > strictly honor batch size. I am just asking if compression batch size > > > message reaches this configured limit, does batch broken down to sub > > batch > > > that is within limit. I would like to minimize the data loss due to > this > > > limit. > > > > > > > > > Thanks, > > > > > > Bhavesh > > > > > > On Mon, Sep 8, 2014 at 9:17 PM, Jun Rao <jun...@gmail.com> wrote: > > > > > > > Are you using compression in the producer? If so, message.max.bytes > > > applies > > > > to the compressed size of a batch of messages. Otherwise, > > > message.max.bytes > > > > applies to the size of each individual message. > > > > > > > > Thanks, > > > > > > > > Jun > > > > > > > > On Wed, Sep 3, 2014 at 3:25 PM, Bhavesh Mistry < > > > mistry.p.bhav...@gmail.com > > > > > > > > > wrote: > > > > > > > > > I am referring to wiki > http://kafka.apache.org/08/configuration.html > > > and > > > > > following parameter control max batch message bytes as far as I > know. > > > > > Kafka Community, please correct me if I am wrong. I do not want to > > > > create > > > > > confusion for Kafka User Community here. Also, if you increase > this > > > > limit > > > > > than you have to set the corresponding limit increase on consumer > > side > > > as > > > > > well (fetch.message.max.bytes). > > > > > > > > > > Since we are using batch async mode, our messages are getting drop > > > > sometime > > > > > if the entire batch bytes exceed this limit so I was asking Kafka > > > > > Developers if any optimal way to determine the batch size based on > > this > > > > > limit to minimize the data loss. Because, entire batch is rejected > by > > > > > brokers. > > > > > > > > > > message.max.bytes 1000000 The maximum size of a message that the > > server > > > > can > > > > > receive. It is important that this property be in sync with the > > maximum > > > > > fetch size your consumers use or else an unruly producer will be > able > > > to > > > > > publish messages too large for consumers to consume. > > > > > > > > > > Thanks, > > > > > > > > > > Bhavesh > > > > > > > > > > > > > > > On Wed, Sep 3, 2014 at 2:59 PM, Alexis Midon < > > > > > alexis.mi...@airbedandbreakfast.com> wrote: > > > > > > > > > > > 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 > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
