diff options
| author | David Arthur <mumrah@gmail.com> | 2013-09-09 00:44:36 -0400 |
|---|---|---|
| committer | David Arthur <mumrah@gmail.com> | 2013-09-09 00:47:28 -0400 |
| commit | f67ad27f72aca077f24fa801a9d2d3075d6d5b60 (patch) | |
| tree | b6c76cfe70429b079a147d6e84952b1f63352dc9 /test/test_integration.py | |
| parent | 40d8e9e550b48755e2f40cfd0877a5b848a3254f (diff) | |
| download | kafka-python-f67ad27f72aca077f24fa801a9d2d3075d6d5b60.tar.gz | |
Auto-adjusting consumer fetch size
Related to #42
Adds new ConsumerFetchSizeTooSmall exception that is thrown when
`_decode_message_set_iter` gets a BufferUnderflowError but has not yet
yielded a message
In this event, SimpleConsumer will increase the fetch size by 1.5 and
continue the fetching loop while _not_ increasing the offset (basically
just retries the request with a larger fetch size)
Once the consumer fetch size has been increased, it will remain
increased while SimpleConsumer fetches from that partition
Diffstat (limited to 'test/test_integration.py')
| -rw-r--r-- | test/test_integration.py | 18 |
1 files changed, 6 insertions, 12 deletions
diff --git a/test/test_integration.py b/test/test_integration.py index 6384b09..bf1acc8 100644 --- a/test/test_integration.py +++ b/test/test_integration.py @@ -8,7 +8,6 @@ import random from kafka import * # noqa from kafka.common import * # noqa from kafka.codec import has_gzip, has_snappy - from .fixtures import ZookeeperFixture, KafkaFixture @@ -757,20 +756,15 @@ class TestConsumer(unittest.TestCase): self.assertEquals(resp.error, 0) self.assertEquals(resp.offset, 10) + # Consumer should still get all of them consumer = SimpleConsumer(self.client, "group1", "test_large_messages") - it = consumer.__iter__() - for i in range(10): - self.assertEquals(messages1[i], it.next().message) - - consumer = SimpleConsumer(self.client, "group2", "test_large_messages", fetch_size_bytes=5120) - it = consumer.__iter__() - for i in range(10): - self.assertEquals(messages1[i], it.next().message) - for i in range(10): - self.assertEquals(messages2[i], it.next().message) + all_messages = messages1 + messages2 + for i, message in enumerate(consumer): + self.assertEquals(all_messages[i], message.message) + self.assertEquals(i, 19) def random_string(l): - s = "".join(random.choice(string.printable) for i in xrange(l)) + s = "".join(random.choice(string.letters) for i in xrange(l)) return s if __name__ == "__main__": |