def _update_matching_pairs(self, raw_pairs, watermark):
        """
        Push the pairs into a order-by-leader-index list,
        and evict the pairs which are out-of-watermark
        """
        for (cid, sid) in raw_pairs:
            #fi: follower index, fe: follower example
            assert cid < self._follower_join_window.size(), \
                    "Leader index[%d] out of range"%cid
            assert sid < self._leader_join_window.size(), \
                    "Follower index[%d] out of range"%(sid)

            example_with_index = self._follower_join_window[cid]
            fi, fe = example_with_index.index, example_with_index.item

            example_with_index = self._leader_join_window[sid]
            li, le = example_with_index.index, example_with_index.item
            if li <= self._leader_restart_index:
                logging.warning("Leader index should be bigger than restart "
                                "index, %d > %d for follower idx %d",
                                li, self._leader_restart_index, fi)
                continue

            if abs(fcc.time_diff(fe.event_time, le.event_time)) > \
               self._max_watermark_delay:
                ### unreachable branch
                logging.info('Pair %s:%s out-of-delay, leader et %d, '
                             'follower et %d', le.example_id, fe.example_id,
                             le.event_time, fe.event_time)
                continue

            # cache the latest leader event
            updated = False
            if fi in self._dedup_by_follower_index:
                old_conv_int = fcc.time_diff(fe.event_time, le.event_time)
                new_conv_int = fcc.time_diff(
                    self._dedup_by_follower_index[fi].event_time, le.event_time)
                if abs(old_conv_int) > abs(new_conv_int):
                    self._dedup_by_follower_index[fi] = \
                            IndexedTime(li, le.event_time)
                    updated = True
            else:
                self._dedup_by_follower_index[fi] = \
                        IndexedTime(li, le.event_time)
                updated = True
            # sort by leader index
            if not updated:
                continue
            self._leader_index_ps.put(_IndexedPair(fe, li, fi))

        matches = []
        while not self._leader_index_ps.empty():
            ip = self._leader_index_ps.get()
            if ip.fe.event_time <= watermark:
                if ip.fi not in self._dedup_by_follower_index:
                    logging.info("Ignore the deleted follower index %d", ip.fi)
                    continue
                indexed_time = self._dedup_by_follower_index[ip.fi]
                if indexed_time.li == ip.li:
                    matches.append(ip)
                    del self._dedup_by_follower_index[ip.fi]
                else:
                    logging.info("Example %s matching leader index %s is"   \
                                 " older than %d", ip.fe.example_id,        \
                                 ip.li, indexed_time.li)
            else:
                self._leader_index_ps.put(ip)
                logging.info('Break dumping, event time %s, watermark %s',
                             ip.fe.event_time, watermark)
                break
        return matches

        Push the pairs into a order-by-leader-index list,
        and evict the pairs which are out-of-watermark
        
fedlearner/data_join/joiner_impl/universal_joiner.py