Log.warn if found a message in kafka topic larger than the maximum fetch size

[jiasheng55]

In kafka ConsumerIterator, there is some codes like this:

// 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))

When "fetch.message.max.bytes" config is smaller than the actual message size in topic, ConsumerIterator will throw an exception to notify user.
But in storm-kafka, there is no such logic. And as a result, if KafkaConfig.fetchSizeBytes is smaller than actual message size, the topology will fetch no data but still be running.
To prevent this situation, we need throw MessageSizeTooLargeException as well.

[abhishekagarwal87]

This would mean spout will get stuck at that particular message and go into infinite worker jvm restarts until config is updated. Isn't it?

[jiasheng55]

Spout will not get stuck and keep running like nothing goes wrong. Users will not know what happened but their topology just stop processing kafka messages.

[abhishekagarwal87]

MessageSizeTooLarge exception will kill the worker JVM (since it is RuntimeException). When worker is re-launched again, the jvm will again be killed when spout reaches the faulty offset. This behavior may not be desired for many users who don't want to stall the worker because of one large message. Can you instead log an Error, ignore the message and proceed further?

Also I didn't fully understand the side effect of not throwing an exception. what do you mean by "topology will fetch no data but still be running"? Will it stop fetching data at all?

[jiasheng55]

Can you instead log an Error, ignore the message and proceed further?

This means data loss to users. I am not sure data loss is acceptable or not. As I mentioned above, ConsumerIterator chooses to throw an exception (MessageTooLargeException, which will cause Kafka Consumer to stop working), so I think maybe it is a good way.

what do you mean by "topology will fetch no data but still be running"? Will it stop fetching data at all?

The spout will keep trying to fetch data, but the response from Kafka contains no valid bytes because of size limit. The side effect of this is that the data in Kafka topic will pile up while users don't know why their storm topology stops processing messages (there is no data to process).

I think you are right that many users don't want to stall the worker because of one large message, but this a result of incorrect config (KafkaConfig.fetchSizeBytes) and if they want to avoid this situation, they need to set a really large size limit at the first time.
Maybe there are better ways to handle this situation, let's figure it out together:)

[revans2]

@abhishekagarwal87 and @Victor-Wong

I am on the fence on this. Having a spout that is stuck forever is really bad, but having it crash, lose data, come back up and repeat the process possibly draining the other partitions feels even worse. I guess if you configured your spouts for at most once processing then you got what you asked for, even if it was shoot yourself in the foot, and storm is fail fast so it fits with that philosophy.

Please at least update the exception message to indicate what the sizes are actually set to. I think this would make life simpler for the user that is in this situation, so they see the error message and it says something like.

Found a message (10,485,760 bytes) that is larger than the maximum fetch size (1,048,576 bytes) in topic myGreatTopic partition 5 at fetch offset 103404502. Increase the fetch size, or decrease the maximum message size the broker will allow and start after this offset."

Another alternative might be to give the user the option to skip messages that are too large, and provide a metric to indicate how many messages/bytes have been skipped because of this.

[jiasheng55]

Found a message (10,485,760 bytes) ...

It seems we can not get the actual size of the message.

I have made some updates according to your advice.

• Add an option "KafkaConfig.skipMsgOverFetchSize" to determine how to handle a large message, default is to skip this message and continue processing.
• Move handling logic out of KafkaUtils. Instead handle it in PartitionManager (non-trident) and TridentKafkaEmitter (trident).

[jiasheng55]

I made some changes, what about logging a warn message instead?
why 'The travis CI build failed' happened? It says storm-core test failed, but I made no changes of storm-core.
@abhishekagarwal87

[erikdw]

Nit: Comma at the front of the line is aesthetically displeasing to me (i.e., ugly). ;-)

More importantly, what is the value in msgs.sizeInBytes() if it's not the message size? i.e., in the review comments you said:

It seems we can not get the actual size of the message.

So I wonder what the value is in msgs.sizeInBytes() then?