tornado.queues consumers are difficult to stop cleanly

While using tornado 5.0.2, the following code will report
ERROR:asyncio:Task was destroyed but it is pending! task: <Task pending coro=<consumer() running at a.py:18> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x10eb7a6a8>()]> cb=[IOLoop.add_future.<locals>.<lambda>() at /usr/local/lib/python3.6/site-packages/tornado/ioloop.py:720]>

this code is copied from http://www.tornadoweb.org/en/stable/queues.html

from tornado import gen
from tornado.ioloop import IOLoop
from tornado.queues import Queue

q = Queue(maxsize=2)

async def consumer():
    async for item in q:
        try:
            print('Doing work on %s' % item)
        finally:
            q.task_done()


async def producer():
    for item in range(1):
        await q.put(item)
        print('Put %s' % item)


async def main():
    # Start consumer without waiting (since it never finishes).
    IOLoop.current().spawn_callback(consumer)
    await producer()     # Wait for producer to put all tasks.
    await q.join()       # Wait for consumer to finish all tasks.
    print('Done')

IOLoop.current().run_sync(main)

Start by running the reproducer from the issue with Tornado 5.0.2 and inspect the tornado.queues consumer behavior around async iteration and shutdown. Done means the example can stop the consumer cleanly after q.join() without reporting a pending destroyed task.