def build_next(self, next_partitions):
        """
        Arguments:
            next_partitions (dict): TopicPartition -> TopicPartitionState

        Returns:
            FetchRequestData
        """
        if self.next_metadata.is_full:
            log.debug("Built full fetch %s for node %s with %s partition(s).",
                self.next_metadata, self.node_id, len(next_partitions))
            self.session_partitions = next_partitions
            return FetchRequestData(next_partitions, None, self.next_metadata)

        prev_tps = set(self.session_partitions.keys())
        next_tps = set(next_partitions.keys())
        log.debug("Building incremental partitions from next: %s, previous: %s", next_tps, prev_tps)
        added = next_tps - prev_tps
        for tp in added:
            self.session_partitions[tp] = next_partitions[tp]
        removed = prev_tps - next_tps
        for tp in removed:
            self.session_partitions.pop(tp)
        altered = set()
        for tp in next_tps & prev_tps:
            if next_partitions[tp] != self.session_partitions[tp]:
                self.session_partitions[tp] = next_partitions[tp]
                altered.add(tp)

        log.debug("Built incremental fetch %s for node %s. Added %s, altered %s, removed %s out of %s",
                  self.next_metadata, self.node_id, added, altered, removed, self.session_partitions.keys())
        to_send = collections.OrderedDict({tp: next_partitions[tp] for tp in next_partitions if tp in (added | altered)})
        return FetchRequestData(to_send, removed, self.next_metadata)

        Arguments:
            next_partitions (dict): TopicPartition -> TopicPartitionState

        Returns:
            FetchRequestData
        
kafka/consumer/fetcher.py