Remove the check for timestamp None in producer, as it's done in RecordBatch anyway.

Minor abc doc fixes.

Author: tvoinarovskyi

kafka/consumer/fetcher.py

@@ -551,12 +551,6 @@ def _unpack_message_set(self, tp, records):
             log.exception('StopIteration raised unpacking messageset: %s', e)
             raise Exception('StopIteration raised unpacking messageset')
 
-        # If unpacking raises AssertionError, it means decompression unsupported
-        # See Issue 1033
-        except AssertionError as e:
-            log.exception('AssertionError raised unpacking messageset: %s', e)
-            raise
-
     def __iter__(self):  # pylint: disable=non-iterator-returned
         return self
 

kafka/producer/kafka.py

@@ -544,8 +544,6 @@ def send(self, topic, value=None, key=None, partition=None, timestamp_ms=None):
             self._ensure_valid_record_size(message_size)
 
             tp = TopicPartition(topic, partition)
-            if timestamp_ms is None:
-                timestamp_ms = int(time.time() * 1000)
             log.debug("Sending (key=%r value=%r) to %s", key, value, tp)
             result = self._accumulator.append(tp, timestamp_ms,
                                               key_bytes, value_bytes,

kafka/record/abc.py

@@ -36,28 +36,21 @@ def checksum(self):
             be the checksum for v0 and v1 and None for v2 and above.
         """
 
-    @abc.abstractproperty
-    def headers(self):
-        """ If supported by version list of key-value tuples, or empty list if
-            not supported by format.
-        """
-
 
 class ABCRecordBatchBuilder(object):
     __metaclass__ = abc.ABCMeta
 
     @abc.abstractmethod
-    def append(self, offset, timestamp, key, value, headers):
+    def append(self, offset, timestamp, key, value):
         """ Writes record to internal buffer.
 
         Arguments:
             offset (int): Relative offset of record, starting from 0
-            timestamp (int): Timestamp in milliseconds since beginning of the
-                epoch (midnight Jan 1, 1970 (UTC))
+            timestamp (int or None): Timestamp in milliseconds since beginning
+                of the epoch (midnight Jan 1, 1970 (UTC)). If omited, will be
+                set to current time.
             key (bytes or None): Key of the record
             value (bytes or None): Value of the record
-            headers (List[Tuple[str, bytes]]): Headers of the record. Header
-                keys can not be ``None``.
 
         Returns:
             (bytes, int): Checksum of the written record (or None for v2 and
@@ -74,10 +67,10 @@ def size_in_bytes(self, offset, timestamp, key, value, headers):
     @abc.abstractmethod
     def build(self):
         """ Close for append, compress if needed, write size and header and
-            return a ready to send bytes object.
+            return a ready to send buffer object.
 
             Return:
-                io.BytesIO: finished batch, ready to send.
+                bytearray: finished batch, ready to send.
         """
 
 
@@ -105,7 +98,7 @@ def __init__(self, buffer):
 
     @abc.abstractmethod
     def size_in_bytes(self):
-        """ Returns the size of buffer.
+        """ Returns the size of inner buffer.
         """
 
     @abc.abstractmethod