summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorDana Powers <dana.powers@gmail.com>2018-02-05 14:02:24 -0800
committerDana Powers <dana.powers@gmail.com>2018-02-05 14:02:24 -0800
commitc6b54036af4aebaca545898fe0d8100a15f93ee8 (patch)
tree8f834926048f56755bbd146b3e97a1370aa6afb4
parent619895d1216cfc3b5da4270d3869cd31b7b55a14 (diff)
downloadkafka-python-fetch_partition_compressed_offset.tar.gz
Fix for PartitionRecords has no attribute message_idx; add testfetch_partition_compressed_offset
-rw-r--r--kafka/consumer/fetcher.py4
-rw-r--r--test/test_fetcher.py23
2 files changed, 24 insertions, 3 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index 6597448..c9bbb97 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -846,6 +846,10 @@ class Fetcher(six.Iterator):
for i, msg in enumerate(messages):
if msg.offset == fetch_offset:
self.message_idx = i
+ break
+ else:
+ self.message_idx = 0
+ self.messages = None
# For truthiness evaluation we need to define __len__ or __nonzero__
def __len__(self):
diff --git a/test/test_fetcher.py b/test/test_fetcher.py
index aa8e9c3..4547222 100644
--- a/test/test_fetcher.py
+++ b/test/test_fetcher.py
@@ -512,12 +512,29 @@ def test_partition_records_offset():
None, None, 'key', 'value', 'checksum', 0, 0)
for i in range(batch_start, batch_end)]
records = Fetcher.PartitionRecords(fetch_offset, None, messages)
- assert records.has_more()
+ assert len(records) > 0
msgs = records.take(1)
assert msgs[0].offset == 123
assert records.fetch_offset == 124
msgs = records.take(2)
assert len(msgs) == 2
- assert records.has_more()
+ assert len(records) > 0
records.discard()
- assert not records.has_more()
+ assert len(records) == 0
+
+
+def test_partition_records_empty():
+ records = Fetcher.PartitionRecords(0, None, [])
+ assert len(records) == 0
+
+
+def test_partition_records_no_fetch_offset():
+ batch_start = 0
+ batch_end = 100
+ fetch_offset = 123
+ tp = TopicPartition('foo', 0)
+ messages = [ConsumerRecord(tp.topic, tp.partition, i,
+ None, None, 'key', 'value', 'checksum', 0, 0)
+ for i in range(batch_start, batch_end)]
+ records = Fetcher.PartitionRecords(fetch_offset, None, messages)
+ assert len(records) == 0