def _update_fetch_positions(self, partitions):
        if (self.config['api_version'] >= (0, 8, 1)
            and self.config['group_id'] is not None):

            # refresh commits for all assigned partitions
            self._coordinator.refresh_committed_offsets_if_needed()

        # then do any offset lookups in case some positions are not known
        self._fetcher.update_fetch_positions(partitions)
Set the fetch position to the committed position (if there is one)
dataset/ETHPy150Open dpkp/kafka-python/kafka/consumer/group.py/KafkaConsumer._update_fetch_positions