(expected_offsets)
Takes in a dictionary of offsets (topics to checkpoint numbers) and ensures they are all available
    in the current kafka feed
Takes in a dictionary of offsets (topics to checkpoint numbers) and ensures they are all available
    in the current kafka feed
Takes in a dictionary of offsets ( topics to checkpoint numbers ) and ensures they are all available in the current kafka feed
def validate_offsets(expected_offsets):
    """
    Takes in a dictionary of offsets (topics to checkpoint numbers) and ensures they are all available
    in the current kafka feed
    """
    if expected_offsets:
        topics = {x[0] for x in expected_offsets.keys()}
        available_offsets = get_multi_topic_first_available_offsets(topics)
        for topic_partition, offset in expected_offsets.items():
            topic, partition = topic_partition
            if topic_partition not in available_offsets:
                raise UnavailableKafkaOffset("Invalid partition '{}' for topic '{}'".format(partition, topic))

            if expected_offsets[topic_partition] < available_offsets[topic_partition]:
                message = (
                    'First available topic offset for {}:{} is {} but needed {}.'
                ).format(topic, partition, available_offsets[topic_partition], expected_offsets[topic_partition])
                raise UnavailableKafkaOffset(message)
def validate_offsets ( expected_offsets ) : if expected_offsets : topics = { x [ 0 ] for x in expected_offsets . keys ( ) } available_offsets = get_multi_topic_first_available_offsets ( topics ) for topic_partition , offset in expected_offsets . items ( ) : topic , partition = topic_partition if topic_partition not in available_offsets : raise UnavailableKafkaOffset ( "Invalid partition '{}' for topic '{}'" . format ( partition , topic ) ) if expected_offsets [ topic_partition ] < available_offsets [ topic_partition ] : message = ( 'First available topic offset for {}:{} is {} but needed {}.' ) . format ( topic , partition , available_offsets [ topic_partition ] , expected_offsets [ topic_partition ] ) raise UnavailableKafkaOffset ( message )