discard leader_epoch data for now

Author: dpkp

kafka/consumer/fetcher.py

@@ -253,10 +253,9 @@ def _retrieve_offsets(self, timestamps, timeout_ms=float("inf")):
         or ``timeout_ms`` passed.
 
         Arguments:
-            timestamps: {TopicPartition: (int, int)} dict with (timestamp, leader_epoch)
-                tuples s to fetch offsets by. Timestamp is -1 for the latest available, and
-                -2 for the earliest available. Otherwise timestamp is treated as epoch milliseconds.
-                Leader epoch is -1 to ignore, otherwise last known epoch value from partition.
+            timestamps: {TopicPartition: int} dict with timestamps to fetch
+                offsets by. -1 for the latest available, -2 for the earliest
+                available. Otherwise timestamp is treated as epoch milliseconds.
 
         Returns:
             {TopicPartition: OffsetAndTimestamp}: Mapping of partition to
@@ -385,7 +384,8 @@ def _append(self, drained, part, max_records, update_offsets):
                     drained[tp].append(record)
 
                 if update_offsets:
-                    self._subscriptions.assignment[tp].position = OffsetAndMetadata(next_offset, '', leader_epoch)
+                    # TODO: save leader_epoch
+                    self._subscriptions.assignment[tp].position = OffsetAndMetadata(next_offset, '', -1)
                 return len(part_records)
 
             else:
@@ -549,7 +549,7 @@ def _send_list_offsets_requests(self, timestamps):
                 return Future().failure(
                     Errors.LeaderNotAvailableError(partition))
             else:
-                leader_epoch = self._client.cluster.leader_epoch_for_partition(partition)
+                leader_epoch = -1
                 timestamps_by_node[node_id][partition] = (timestamp, leader_epoch)
 
         # Aggregate results until we have all

kafka/coordinator/consumer.py

@@ -835,7 +835,8 @@ def _handle_offset_fetch_response(self, future, response):
                 elif offset >= 0:
                     # record the position with the offset
                     # (-1 indicates no committed offset to fetch)
-                    offsets[tp] = OffsetAndMetadata(offset, metadata, leader_epoch)
+                    # TODO: save leader_epoch
+                    offsets[tp] = OffsetAndMetadata(offset, metadata, -1)
                 else:
                     log.debug("Group %s has no committed offset for partition"
                               " %s", self.group_id, tp)

test/test_fetcher.py

@@ -244,8 +244,8 @@ def send_side_effect(node_id, timestamps):
         else:
             second_future = f
     assert req_by_node == {
-        0: {tp1: (0, 0), tp3: (0, 0)},
-        1: {tp2: (0, 0), tp4: (0, 0)}
+        0: {tp1: (0, -1), tp3: (0, -1)},
+        1: {tp2: (0, -1), tp4: (0, -1)}
     }
 
     # We only resolved 1 future so far, so result future is not yet ready