revert last_offset_from_record_batch

Author: dpkp

kafka/consumer/fetcher.py

@@ -417,7 +417,6 @@ def _unpack_records(self, tp, records):
                 try:
                     batch_offset = batch.base_offset + batch.last_offset_delta
                     leader_epoch = batch.leader_epoch
-                    self._subscriptions.assignment[tp].last_offset_from_record_batch = batch_offset
                     # Control batches have a single record indicating whether a transaction
                     # was aborted or committed.
                     # When isolation_level is READ_COMMITTED (currently unsupported)
@@ -643,16 +642,6 @@ def _create_fetch_requests(self):
         for partition in self._fetchable_partitions():
             node_id = self._client.cluster.leader_for_partition(partition)
 
-            # advance position for any deleted compacted messages if required
-            if self._subscriptions.assignment[partition].last_offset_from_record_batch:
-                next_offset_from_batch_header = self._subscriptions.assignment[partition].last_offset_from_record_batch + 1
-                if next_offset_from_batch_header > self._subscriptions.assignment[partition].position.offset:
-                    log.debug(
-                        "Advance position for partition %s from %s to %s (last record batch location plus one)"
-                        " to correct for deleted compacted messages and/or transactional control records",
-                        partition, self._subscriptions.assignment[partition].position.offset, next_offset_from_batch_header)
-                    self._subscriptions.assignment[partition].position = OffsetAndMetadata(next_offset_from_batch_header, '', -1)
-
             position = self._subscriptions.assignment[partition].position
 
             # fetch if there is a leader and no in-flight requests

kafka/consumer/subscription_state.py

@@ -382,9 +382,6 @@ def __init__(self):
         self._position = None # OffsetAndMetadata exposed to the user
         self.highwater = None
         self.drop_pending_record_batch = False
-        # The last message offset hint available from a record batch with
-        # magic=2 which includes deleted compacted messages
-        self.last_offset_from_record_batch = None
 
     def _set_position(self, offset):
         assert self.has_valid_position, 'Valid position required'
@@ -400,7 +397,6 @@ def await_reset(self, strategy):
         self.awaiting_reset = True
         self.reset_strategy = strategy
         self._position = None
-        self.last_offset_from_record_batch = None
         self.has_valid_position = False
 
     def seek(self, offset):
@@ -409,7 +405,6 @@ def seek(self, offset):
         self.reset_strategy = None
         self.has_valid_position = True
         self.drop_pending_record_batch = True
-        self.last_offset_from_record_batch = None
 
     def pause(self):
         self.paused = True