(
        self,
        since: Optional[Dict[TopicPartition, int]],
        forever: bool,
    )
``since`` must be a dictionary of topic partition offsets, or None
``since`` must be a dictionary of topic partition offsets, or None
since must be a dictionary of topic partition offsets or None
def iter_changes(
        self,
        since: Optional[Dict[TopicPartition, int]],
        forever: bool,
    ) -> Iterator[Change]:
        """
        ``since`` must be a dictionary of topic partition offsets, or None
        """
        timeout = MAX_TIMEOUT if forever else MIN_TIMEOUT
        start_from_latest = since is None
        reset = 'largest' if start_from_latest else 'smallest'
        self._init_consumer(timeout, auto_offset_reset=reset)

        since = self._filter_offsets(since)
        # a special value of since=None will start from the end of the change stream
        if not isinstance(since, (dict, type(None))):
            raise ValueError(f"Expected None or a topic offset dictionary. Got {since!r}")
        if since == {}:
            raise ValueError('Topic partition offsets not found')

        if not start_from_latest:
            if self.strict:
                validate_offsets(since)

            checkpoint_topics = {tp[0] for tp in since}
            extra_topics = checkpoint_topics - set(self._topics)
            if extra_topics:
                raise ValueError("'since' contains extra topics: {}".format(list(extra_topics)))

            self._processed_topic_offsets = copy(since)

            # Tell the consumer to start from offsets that were passed in
            for topic_partition, offset in since.items():
                self.consumer.seek(TopicPartition(topic_partition[0], topic_partition[1]), int(offset))

        try:
            for message in self.consumer:
                self._processed_topic_offsets[(message.topic, message.partition)] = message.offset
                yield change_from_kafka_message(message)
        except StopIteration:
            # no need to do anything since this is just telling us we've reached the end of the feed
            pass
def iter_changes ( self , since : Optional [ Dict [ TopicPartition , int ] ] , forever : bool , ) -> Iterator [ Change ] : timeout = MAX_TIMEOUT if forever else MIN_TIMEOUT start_from_latest = since is None reset = 'largest' if start_from_latest else 'smallest' self . _init_consumer ( timeout , auto_offset_reset = reset ) since = self . _filter_offsets ( since ) # a special value of since=None will start from the end of the change stream if not isinstance ( since , ( dict , type ( None ) ) ) : raise ValueError ( f"Expected None or a topic offset dictionary. Got {since!r}" ) if since == { } : raise ValueError ( 'Topic partition offsets not found' ) if not start_from_latest : if self . strict : validate_offsets ( since ) checkpoint_topics = { tp [ 0 ] for tp in since } extra_topics = checkpoint_topics - set ( self . _topics ) if extra_topics : raise ValueError ( "'since' contains extra topics: {}" . format ( list ( extra_topics ) ) ) self . _processed_topic_offsets = copy ( since ) # Tell the consumer to start from offsets that were passed in for topic_partition , offset in since . items ( ) : self . consumer . seek ( TopicPartition ( topic_partition [ 0 ] , topic_partition [ 1 ] ) , int ( offset ) ) try : for message in self . consumer : self . _processed_topic_offsets [ ( message . topic , message . partition ) ] = message . offset yield change_from_kafka_message ( message ) except StopIteration : # no need to do anything since this is just telling us we've reached the end of the feed pass