diff options
| author | Dana Powers <dana.powers@gmail.com> | 2016-07-17 11:44:01 -0700 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2016-07-17 11:44:01 -0700 |
| commit | 68c8a4a276237488fba1495fcec43b572a80289c (patch) | |
| tree | 4b61fc55c471bf4095cc72777c9a4a69a316dd9b /kafka/consumer/fetcher.py | |
| parent | bcbc0c4e9af01009dd1750aa7d939e2489ddeb73 (diff) | |
| download | kafka-python-68c8a4a276237488fba1495fcec43b572a80289c.tar.gz | |
KAFKA-3196: Add checksum and size to RecordMetadata and ConsumerRecord (#770 / #594)
Diffstat (limited to 'kafka/consumer/fetcher.py')
| -rw-r--r-- | kafka/consumer/fetcher.py | 11 |
1 files changed, 8 insertions, 3 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index 11cd7ef..f780fb2 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -20,7 +20,8 @@ log = logging.getLogger(__name__) ConsumerRecord = collections.namedtuple("ConsumerRecord", - ["topic", "partition", "offset", "timestamp", "timestamp_type", "key", "value"]) + ["topic", "partition", "offset", "timestamp", "timestamp_type", + "key", "value", "checksum", "serialized_key_size", "serialized_value_size"]) class NoOffsetForPartitionError(Errors.KafkaError): @@ -410,13 +411,17 @@ class Fetcher(six.Iterator): key, value = self._deserialize(inner_msg) yield ConsumerRecord(tp.topic, tp.partition, inner_offset, inner_timestamp, msg.timestamp_type, - key, value) + key, value, inner_msg.crc, + len(inner_msg.key) if inner_msg.key is not None else -1, + len(inner_msg.value) if inner_msg.value is not None else -1) else: key, value = self._deserialize(msg) yield ConsumerRecord(tp.topic, tp.partition, offset, msg.timestamp, msg.timestamp_type, - key, value) + key, value, msg.crc, + len(msg.key) if msg.key is not None else -1, + len(msg.value) if msg.value is not None else -1) # If unpacking raises StopIteration, it is erroneously # caught by the generator. We want all exceptions to be raised |