Merge pull request #73 from Yelp/u/georgios/KAFKA-36342_fix_control_batches_bug

Fix control batch bug

Author: gkousouris

kafka/consumer/fetcher.py

@@ -821,11 +821,12 @@ def _parse_fetched_data(self, completed_fetch):
                               " offset %d to buffered record list", tp,
                               position)
                     unpacked = list(self._unpack_message_set(tp, records))
-                    parsed_records = self.PartitionRecords(fetch_offset, tp, unpacked)
-                    last_offset = unpacked[-1].offset
-                    self._sensors.records_fetch_lag.record(highwater - last_offset)
-                    num_bytes = records.valid_bytes()
-                    records_count = len(unpacked)
+                    if unpacked:
+                        parsed_records = self.PartitionRecords(fetch_offset, tp, unpacked)
+                        last_offset = unpacked[-1].offset
+                        self._sensors.records_fetch_lag.record(highwater - last_offset)
+                        num_bytes = records.valid_bytes()
+                        records_count = len(unpacked)
                 elif records.size_in_bytes() > 0:
                     # we did not read a single message from a non-empty
                     # buffer because that message's size is larger than

kafka/version.py

@@ -1 +1 @@
-__version__ = '1.4.7.post3'
+__version__ = '1.4.7.post4'