Connection to broker shouldn't be terminated when kafka returns `MessageSizeTooLarge` error

[jasantunes]

I recently came across the MessageSizeTooLarge error response from kafka, which I think it's not being properly handled by the ruby-kafka, at least in the async call from the producer. Currently it seems to be disconnecting the producer, thus triggering a bunch on other issues and ultimately 500s. The synchronized call seems to be handling the error fine.

• Version of Ruby: 2.2.3, 2.3.3
• Version of Kafka: 0.10.1.0
• Version of ruby-kafka: 0.3, master

Steps to reproduce

When calling it synchronously, the exception is raised properly without affecting the kafka connection:

2.2.3 (main):0 > k = Kafka.new(logger: Logger.new(STDOUT), seed_brokers: [ 'localhost'], client_id: "my-application")
2.2.3 (main):0 > k.deliver_message("A"*1500000, topic: "development_events")
I, [2017-05-24T17:36:14.573805 #70030]  INFO -- : New topics added to target list: development_events
I, [2017-05-24T17:36:14.573922 #70030]  INFO -- : Fetching cluster metadata from kafka://localhost:9092
D, [2017-05-24T17:36:14.574013 #70030] DEBUG -- : Opening connection to localhost:9092 with client id my-application...
D, [2017-05-24T17:36:14.576325 #70030] DEBUG -- : Sending request 1 to localhost:9092
D, [2017-05-24T17:36:14.576523 #70030] DEBUG -- : Waiting for response 1 from localhost:9092
D, [2017-05-24T17:36:14.580413 #70030] DEBUG -- : Received response 1 from localhost:9092
I, [2017-05-24T17:36:14.580479 #70030]  INFO -- : Discovered cluster metadata; nodes: localhost:9092 (node_id=0)
D, [2017-05-24T17:36:14.580559 #70030] DEBUG -- : Closing socket to localhost:9092
D, [2017-05-24T17:36:14.580825 #70030] DEBUG -- : Current leader for development_events/0 is node localhost:9092 (node_id=0)
I, [2017-05-24T17:36:14.580913 #70030]  INFO -- : Sending 1 messages to localhost:9092 (node_id=0)
D, [2017-05-24T17:36:14.580956 #70030] DEBUG -- : Opening connection to localhost:9092 with client id my-application...
D, [2017-05-24T17:36:14.581860 #70030] DEBUG -- : Sending request 1 to localhost:9092
D, [2017-05-24T17:36:14.587700 #70030] DEBUG -- : Waiting for response 1 from localhost:9092
D, [2017-05-24T17:36:14.597992 #70030] DEBUG -- : Received response 1 from localhost:9092
Kafka::MessageSizeTooLarge: Kafka::MessageSizeTooLarge
from /Users/joao/dev/ruby-kafka/lib/kafka/protocol.rb:50:in `handle_error'
(...)

However, when calling the async version, the exception gets raised and the connection to the broker is terminated:

2.2.3 (main):0 > k = Kafka.new(logger: Logger.new(STDOUT), seed_brokers: [ 'localhost'], client_id: "my-application")
2.2.3 (main):0 > p = k.async_producer
2.2.3 (main):0 > p.produce("A"*1500000, topic: "development_events")
=> nil
2.2.3 (main):0 > p.deliver_messages
I, [2017-05-24T17:46:22.724993 #70268]  INFO -- : New topics added to target list: development_events
I, [2017-05-24T17:46:22.725138 #70268]  INFO -- : Fetching cluster metadata from kafka://localhost:9092
D, [2017-05-24T17:46:22.725322 #70268] DEBUG -- : Opening connection to localhost:9092 with client id my-application...
=> nil
2.2.3 (main):0 > D, [2017-05-24T17:46:22.727822 #70268] DEBUG -- : Sending request 1 to localhost:9092
D, [2017-05-24T17:46:22.728048 #70268] DEBUG -- : Waiting for response 1 from localhost:9092
D, [2017-05-24T17:46:22.731578 #70268] DEBUG -- : Received response 1 from localhost:9092
I, [2017-05-24T17:46:22.731628 #70268]  INFO -- : Discovered cluster metadata; nodes: localhost:9092 (node_id=0)
D, [2017-05-24T17:46:22.731651 #70268] DEBUG -- : Closing socket to localhost:9092
D, [2017-05-24T17:46:22.731870 #70268] DEBUG -- : Current leader for development_events/0 is node localhost:9092 (node_id=0)
I, [2017-05-24T17:46:22.731937 #70268]  INFO -- : Sending 1 messages to localhost:9092 (node_id=0)
D, [2017-05-24T17:46:22.731989 #70268] DEBUG -- : Opening connection to localhost:9092 with client id my-application...
D, [2017-05-24T17:46:22.732612 #70268] DEBUG -- : Sending request 1 to localhost:9092
D, [2017-05-24T17:46:22.738458 #70268] DEBUG -- : Waiting for response 1 from localhost:9092
D, [2017-05-24T17:46:22.747448 #70268] DEBUG -- : Received response 1 from localhost:9092
I, [2017-05-24T17:46:22.747647 #70268]  INFO -- : Disconnecting broker 0
D, [2017-05-24T17:46:22.747681 #70268] DEBUG -- : Closing socket to localhost:9092
Error: Kafka::MessageSizeTooLarge
/Users/joao/dev/ruby-kafka/lib/kafka/protocol.rb:50:in `handle_error'(...)

Expected outcome

When using the async call, the connection shouldn't be terminated when kafka returns MessageSizeTooLarge error. Behavior should be similar to the non-async version.

Actual outcome

When using the async call, the connection gets terminated when kafka returns MessageSizeTooLarge error.

[dasch]

The problem is that the only approach the async producer has when dealing with errors is to retry – but a message that is too large won't succeed when retried.

I've thought about this previously, but couldn't come up with a really good error handling API that

1. was simple to use,
2. provided actionable information to the user,
3. and guarded against data loss.

Especially 3) is tricky. It would be quite easy to simply drop messages that are too large, but that would mean losing data.

This might be okay since the async producer doesn't guarantee delivery in the first place, but it would be important to instrument the occurrences and provide logging and monitoring.

Note that it's not possible to simply raise an exception, since by the time the producer gets the error, the caller will be doing other stuff – the producer runs in the background. It's also not possible to guard against too large messages in the ruby-kafka code, since that's configurable in the Kafka brokers.

I'm open to proposals.

[SimSimY]

IMO there should be at least an option skip the failing message or to pass a block that will handle this case (so we can export them to some storage and handle them later)

In our case, we have api endpoint that receives events and pushes them to kafka, while we can tolerate some data loss ( around ~.001% in our case) every time there is Kafka::MessageSizeTooLarge error, the whole worker (we are using puma) is forced to reload, causes bigger data loss.

[dasch]

I would be interested in knowing what the JVM client does in these cases.

[jasantunes]

I like @SimSimY 's suggestion of the error handling block – stoplight uses this approach: https://github.com/orgsync/stoplight#custom-errors

A less elegant solution could be to request the broker the message max size (broker and topic) at the initialization and raise MessageSizeTooLarge if the client tries to call produce with a too large message w/o even trying to actually send the message.

[dasch]

I'm not too happy about using blocks to set policies – I think it's rather inelegant.

If only I could determine the max message size programmatically! Unfortunately, there is to my knowledge no such API in Kafka. Rejecting too large messages in #produce would solve the problem nicely.

[jasantunes]

Yes, getting the max message (per topic) might be unfeasible – I think it's managed by the zookeeper cluster...

[jasantunes]

A hacky (partial) solution: ruby-kafka could track the min of the messages that caused MessageSizeTooLarge per topic, i.e., an approximate of the max for each topic, max*(topic): any messages to be produced larger than max*(topic) will immediately raise a MessageSizeTooLarge without actually trying to send them to kafka.

Caveats:

• requires an additional key/value store to cache max*(topic), such as local mem or Redis
• will lose data on every message that is larger than the actual max, but still smaller than max*(topic)

Pros:

• no more terminated connections on async produce – exceptions would be masked and used to update max*(topic)
• using a shared Redis instance to store max*(topic) would speed up convergence and minimize data loss when using multiple kafka producers

Although IMHO I still think providing a handler block would be a better approach.

[dasch]

I'm not sure that fixes the problem in a complete way – you'll still silently lose the first message to go over the max size at least.

[jasantunes]

@dasch: Not a handler block but a handler object that implements a callback: #358

[film42]

This shouldn't be a problem anymore after #432