code: def consumer_commit_for_times(consumer, partition_to_offset, atomic=False):
      
      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)
docs: 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.