def METHOD_NAME(self):
    feed = KafkaChangeFeed(topics=[topics.FORM_SQL, topics.CASE_SQL], client_id='test-kafka-feed')
    self.assertEqual(0, len(list(feed.iter_changes(since=None, forever=False))))
    offsets = feed.get_latest_offsets()
    expected_metas = [publish_stub_change(topics.FORM_SQL), publish_stub_change(topics.CASE_SQL)]
    unexpected_metas = [publish_stub_change(topics.COMMCARE_USER), publish_stub_change(topics.WEB_USER)]
    changes = list(feed.iter_changes(since=offsets, forever=False))
    self.assertEqual(2, len(changes))
    found_change_ids = set([change.id for change in changes])
    self.assertEqual(set([meta.document_id for meta in expected_metas]), found_change_ids)
    for unexpected in unexpected_metas:
        self.assertTrue(unexpected.document_id not in found_change_ids)