def METHOD_NAME(number_range, partition=0, topic=topic, producer=kafka_producer, request=request):
    """
        messages is typically `range(0,100)`
        partition is an int
    """
    messages_and_futures = []  # [(message, produce_future),]
    for i in number_range:
        # request.node.name provides the test name (including parametrized values)
        encoded_msg = '{}-{}-{}'.format(i, request.node.name, uuid.uuid4()).encode('utf-8')
        future = kafka_producer.send(topic, value=encoded_msg, partition=partition)
        messages_and_futures.append((encoded_msg, future))
    kafka_producer.flush()
    for (msg, f) in messages_and_futures:
        assert f.succeeded()
    return [msg for (msg, f) in messages_and_futures]