Handle case where ListOffset returns an empty array of offsets (#78)

There was an assumption made that ListOffset always returns at least one offset. However, in the case where the timestamp requested is out of range (or for brokers that do not have a time index, when the offset is in the first segment) where an empty array is returned. This updates the TopicOffsets model to handle that case and insert a -1 instead to indicate that no offset was available. I also added a test specifically for that case.
linkedin · Sep 29, 2017 · 1031313 · 1031313
1 parent 3609630
commit 1031313
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 2 deletions.
diff --git a/kafka/tools/models/topic.py b/kafka/tools/models/topic.py
@@ -92,7 +92,11 @@ def set_offsets_from_list(self, partitions):
         """
         for partition in partitions:
             raise_if_error(OffsetError, partition['error'])
-            self.partitions[partition['partition']] = partition['offsets'][0]
+            if len(partition['offsets']) > 0:
+                self.partitions[partition['partition']] = partition['offsets'][0]
+            else:
+                # We received no offsets back, so we'll just return -1 to indicate that
+                self.partitions[partition['partition']] = -1
 
     def set_offsets_from_fetch(self, partitions):
         """

diff --git a/tests/tools/client/fixtures.py b/tests/tools/client/fixtures.py
@@ -139,6 +139,16 @@ def list_offset():
                                                                          'offsets': [8904]}]}]})
 
 
+def list_offset_none():
+    return ListOffsetV0Response({'responses': [{'topic': 'topic1',
+                                                'partition_responses': [{'partition': 0,
+                                                                         'error': 0,
+                                                                         'offsets': [4829]},
+                                                                        {'partition': 1,
+                                                                         'error': 0,
+                                                                         'offsets': []}]}]})
+
+
 def list_offset_error():
     return ListOffsetV0Response({'responses': [{'topic': 'topic1',
                                                 'partition_responses': [{'partition': 0,

diff --git a/tests/tools/models/test_topic_and_partition.py b/tests/tools/models/test_topic_and_partition.py
@@ -1,6 +1,6 @@
 import unittest
 
-from tests.tools.client.fixtures import list_offset, list_offset_error, offset_fetch, offset_fetch_error
+from tests.tools.client.fixtures import list_offset, list_offset_none, list_offset_error, offset_fetch, offset_fetch_error
 
 from kafka.tools.exceptions import OffsetError
 from kafka.tools.models.cluster import Cluster
@@ -271,6 +271,15 @@ def test_set_offsets_from_list(self):
         assert offsets.partitions[0] == 4829
         assert offsets.partitions[1] == 8904
 
+    def test_set_offsets_empty_from_list(self):
+        topic = Topic('topic1', 2)
+        offsets = TopicOffsets(topic)
+        response = list_offset_none()
+
+        offsets.set_offsets_from_list(response['responses'][0]['partition_responses'])
+        assert offsets.partitions[0] == 4829
+        assert offsets.partitions[1] == -1
+
     def test_set_offsets_from_list_error(self):
         topic = Topic('topic1', 2)
         offsets = TopicOffsets(topic)