def METHOD_NAME(self, **options):
    topic = options["topic"]
    doc_ids = options.get("doc_ids")
    if doc_ids:
        doc_ids = doc_ids.split(",")
    if options.get("doc_id_file"):
        if doc_ids:
            raise CommandError("Can not supply doc IDs via file and via command line")
        with open(options["doc_id_file"], "r") as f:
            doc_ids = [line.strip() for line in f.readlines()]
    if doc_ids:
        # encode to match kafka message key
        doc_ids = {str(doc_id).encode() for doc_id in doc_ids}
    doc_type = options.get("doc_type")
    doc_subtype = options.get("doc_subtype")
    start, end = options.get("start_date"), options.get("end_date")
    info = [f"Searching the '{topic}' Kafka topic for documents matching:"]
    if doc_ids:
        info.append(f"\t doc_ids (count): {len(doc_ids)}")
    if doc_type:
        info.append(f"\t        doc_type: {doc_type}")
    if doc_subtype:
        info.append(f"\t     doc_subtype: {doc_subtype}")
    if start:
        info.append(f"\t published after: {start}")
    if end:
        info.append(f"\tpublished before: {end}")
    info = "\n".join(info)
    self.stderr.write(f"{info}\n\n")
    partitions = [
        TopicPartition(topic, partition)
        for topic, partition in get_multi_topic_first_available_offsets([topic])
    ]
    consumer = get_consumer()
    consumer.assign(partitions)
    if start:
        self.stderr.write(f"Searching for best offsets to start based on start date: {start}\n")
        offsets = get_offsets(partitions, start)
        for partition in partitions:
            consumer.seek(partition, offsets[partition])
    else:
        consumer.seek_to_beginning(partitions)
    count = 0
    last_progress = None
    buffer = []
    for message in consumer:
        count += 1
        metadata = None
        if count % 1000 == 0:
            if buffer:
                self.stdout.writelines(buffer)
                self.stdout.flush()
                buffer = []
            metadata = change_from_kafka_message(message).metadata
            timestamp = metadata.publish_timestamp
            if not last_progress or (timestamp - last_progress).total_seconds() > (3600 * 2):
                self.stderr.write(f"\nExamined {count} changes. Current point: {timestamp}\n")
                last_progress = timestamp
            if end and metadata.publish_timestamp > end:
                break
        if doc_ids and message.key not in doc_ids:
            continue
        if not metadata:
            change = change_from_kafka_message(message)
            metadata = change.metadata
        if doc_type and metadata.document_type != doc_type:
            continue
        if doc_subtype and metadata.document_subtype != doc_subtype:
            continue
        if end and metadata.publish_timestamp > end:
            break
        output = metadata.to_json()
        output["partition"] = message.partition
        output["offset"] = message.offset
        buffer.append(f"{json.dumps(output)}\n")
    if buffer:
        self.stdout.writelines(buffer)
        self.stdout.flush()