[KAFKA-994] High level consumer doesn't throw an exception when the message it is trying to fetch exceeds the configured fetch size

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.8.0
• Fix Version/s: 0.8.0
• Component/s: consumer
• Assignee: Sam Meder
• Reporter: Sam Meder
• Created: 01/Aug/13 13:59
• Updated: 03/Aug/13 16:49
• Resolved: 02/Aug/13 04:35

Description

The high level consumer code is supposed to throw an exception when it encounters a message that exceeds its configured max message size. The relevant code form ConsumerIterator.scala is:

// if we just updated the current chunk and it is empty that means the fetch size is too small!
if(currentDataChunk.messages.validBytes == 0)
throw new MessageSizeTooLargeException("Found a message larger than the maximum fetch size of this consumer on topic " +
"%s partition %d at fetch offset %d. Increase the fetch size, or decrease the maximum message size the broker will allow."
.format(currentDataChunk.topicInfo.topic, currentDataChunk.topicInfo.partitionId, currentDataChunk.fetchOffset))
}

The problem is that KAFKA-846 changed PartitionTopicInfo.enqueue:

def enqueue(messages: ByteBufferMessageSet) {

• val size = messages.sizeInBytes
  + val size = messages.validBytes
  if(size > 0) {

i.e. chunks that contain messages that are too big (validBytes = 0) will never even be enqueued, so won't ever hit the too-large message check in ConsumerIterator...

I've attached a patch that passes our tests...