(consumer, tp, target_date, margin)
return check_offset
Creates a function that is used to evaluate a Kafka partition offset to determine
    if the offset is at, above or below the desired location (based on the target_date).


    :param consumer: Kafka consumer
    :param tp: TuplePartition that is being searched
    :param target_date: change published date that is being searched for
    :param margin: margin or error that should be tolerated (in seconds)
    :returns: A function that can be used as a predicate to the ``_search`` function.
Creates a function that is used to evaluate a Kafka partition offset to determine
    if the offset is at, above or below the desired location (based on the target_date).
Creates a function that is used to evaluate a Kafka partition offset to determine if the offset is at above or below the desired location ( based on the target_date ) .
def get_offset_check_function(consumer, tp, target_date, margin):
    """
    Creates a function that is used to evaluate a Kafka partition offset to determine
    if the offset is at, above or below the desired location (based on the target_date).


    :param consumer: Kafka consumer
    :param tp: TuplePartition that is being searched
    :param target_date: change published date that is being searched for
    :param margin: margin or error that should be tolerated (in seconds)
    :returns: A function that can be used as a predicate to the ``_search`` function.

    """

    def check_offset(offset, consumer=consumer, tp=tp, target_date=target_date, margin=margin):
        consumer.seek(tp, offset)
        message = next(consumer)
        change = change_from_kafka_message(message)
        diff = (change.metadata.publish_timestamp - target_date).total_seconds()
        if diff < 0 and abs(diff) < margin:
            return 0
        return diff

    return check_offset
def get_offset_check_function ( consumer , tp , target_date , margin ) : def check_offset ( offset , consumer = consumer , tp = tp , target_date = target_date , margin = margin ) : consumer . seek ( tp , offset ) message = next ( consumer ) change = change_from_kafka_message ( message ) diff = ( change . metadata . publish_timestamp - target_date ) . total_seconds ( ) if diff < 0 and abs ( diff ) < margin : return 0 return diff return check_offset