def reset_offsets_if_needed(self):
        """Reset offsets for the given partitions using the offset reset strategy.

        Arguments:
            partitions ([TopicPartition]): the partitions that need offsets reset

        Returns:
            bool: True if any partitions need reset; otherwise False (no reset pending)

        Raises:
            NoOffsetForPartitionError: if no offset reset strategy is defined
            KafkaTimeoutError if timeout_ms provided
        """
        # Raise exception from previous offset fetch if there is one
        exc, self._cached_list_offsets_exception = self._cached_list_offsets_exception, None
        if exc:
            raise exc

        partitions = self._subscriptions.partitions_needing_reset()
        if not partitions:
            return False
        log.debug('Resetting offsets for %s', partitions)

        offset_resets = dict()
        for tp in partitions:
            ts = self._subscriptions.assignment[tp].reset_strategy
            if ts:
                offset_resets[tp] = ts

        self._reset_offsets_async(offset_resets)
        return True
Reset offsets for the given partitions using the offset reset strategy.

        Arguments:
            partitions ([TopicPartition]): the partitions that need offsets reset

        Returns:
            bool: True if any partitions need reset; otherwise False (no reset pending)

        Raises:
            NoOffsetForPartitionError: if no offset reset strategy is defined
            KafkaTimeoutError if timeout_ms provided
        
kafka/consumer/fetcher.py