Consumer memory leak in idle state (#629)

* hotfix/ memory leak in comsumer getmany

* hotfix/ Added changes in change log

* hotfix/ Fixed case where pending future set is empty

* hotfix/ renamed pending future variable

CHANGES.rst

@@ -2,6 +2,8 @@
 Changelog
 =========
 
+628.bugfix
+Fix memory leak in kafka consumer when consumer is in idle state not consuming any message
 
 0.6.0 (2020-05-15)
 ==================

aiokafka/consumer/fetcher.py

@@ -1097,10 +1097,13 @@ async def fetched_records(self, partitions, timeout=0, max_records=None):
                 return drained
 
             waiter = self._create_fetch_waiter()
-            done, _ = await asyncio.wait(
+            done, pending = await asyncio.wait(
                 [waiter], timeout=timeout, loop=self._loop)
 
             if not done or self._closed:
+                if pending:
+                    fut = pending.pop()
+                    fut.cancel()
                 return {}
 
             if waiter.done():