mumrah/kafka-python — pull request #88 by rdiomar (2014-01-07 00:17:16+00:00)
Title: Various changes/fixes

This set of changes addresses several bugs and issues, and fixes most of the tests. It also changes the behavior of SimpleConsumer in a few ways. Here is a summary of all the changes, mostly copied from the major commits:

* Allow customizing socket timeouts.
Previously, if you try to consume a message with a timeout greater than 10 seconds, but you don't receive data in those 10 seconds, a socket.timeout exception is raised. This allows a higher socket timeout to be set, or even None for no timeout.

* Read the correct number of bytes from kafka.
According to the protocol documentation, the 4 byte integer at the beginning of a response represents the size of the payload only, not including those bytes. See the Kafka docs [here](https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-CommonRequestandResponseStructure).

* Guarantee reading the expected number of bytes from the socket every time.

* Remove bufsize from client and conn.
Since bufsize in the socket is not actually enforced, but it is used by the consumer when creating requests, moving it there until a better solution is implemented

* SimpleConsumer flow changes:
* Combine partition fetch requests into a single request
* Put the messages received in a queue and update offsets
* Grab as many messages from the queue as requested
* When the queue is empty, request more
* timeout param for get_messages() is the actual timeout for getting those messages
* Based on https://github.com/mumrah/kafka-python/pull/74 -
don't increase min_bytes if the consumer fetch buffer size is too small.
* Notes:
* Previously, when querying each partition separately, it was possible to block waiting for messages on partition 0 even if there are new ones in partition 1. These changes allow us to block while aiting for messages on all partitions, and reduce total number of kafka requests.
* Use Queue.Queue for single proc Queue instead of already imported multiprocessing.Queue because the latter doesn't seem to guarantee immediate availability of items after a put.

* Remove SimpleConsumer queue size limit since it can cause the iterator to block forever if it's reached.

* Fix error handling in client.

* Add an optional limit to how much the consumer fetch buffer size can grow.

* Add and partially fix integration tests. TestFailover tests still fail intermittently.

* Add and fix unit tests.
This is pretty much a rewrite. The tests that involve offset requests/responses are not implemented since that API is not supported in kafka 0.8 yet. Only kafka.codec and kafka.protocol are currently tested, so there is more work to be done here.

* Style fixes and comments.