def consumer_commit_for_times(consumer, partition_to_offset, atomic=False):
    """Commits offsets to Kafka using the given KafkaConsumer and offsets, a mapping
    of TopicPartition to Unix Epoch milliseconds timestamps.

    Arguments:
        consumer (KafkaConsumer): an initialized kafka-python consumer.
        partitions_to_offset (dict TopicPartition: OffsetAndTimestamp): Map of TopicPartition to OffsetAndTimestamp. Return value of offsets_for_times.
        atomic (bool): Flag to specify whether the commit should fail if offsets are not found for some
            TopicPartition: timestamp pairs.
    """
    no_offsets = set()
    for tp, offset in six.iteritems(partition_to_offset):
        if offset is None:
            logging.error(
                "No offsets found for topic-partition {tp}. Either timestamps not supported"
                " for the topic {tp}, or no offsets found after timestamp specified, or there is no"
                " data in the topic-partition.".format(tp=tp),
            )
            no_offsets.add(tp)
    if atomic and len(no_offsets) > 0:
        logging.error(
            "Commit aborted; offsets were not found for timestamps in"
            " topics {}".format(",".join([str(tp) for tp in no_offsets])),
        )
        return

    offsets_metadata = {
        tp: OffsetAndMetadata(partition_to_offset[tp].offset, metadata=None)
        for tp in six.iterkeys(partition_to_offset) if tp not in no_offsets
    }

    if len(offsets_metadata) != 0:
        consumer.commit(offsets_metadata)
Commits offsets to Kafka using the given KafkaConsumer and offsets, a mapping
    of TopicPartition to Unix Epoch milliseconds timestamps.

    Arguments:
        consumer (KafkaConsumer): an initialized kafka-python consumer.
        partitions_to_offset (dict TopicPartition: OffsetAndTimestamp): Map of TopicPartition to OffsetAndTimestamp. Return value of offsets_for_times.
        atomic (bool): Flag to specify whether the commit should fail if offsets are not found for some
            TopicPartition: timestamp pairs.