Only commit offsets when there's a batch
@@ -66,10 +66,10 @@ async def consume(kafka_topic):
if tasks:
log.info('Processing image batch of size {}'.format(len(tasks)))
total += len(tasks)
+ await asyncio.gather(*tasks)
total_time = (dt.datetime.now() - start).total_seconds()
log.info(f'Last batch took {total_time}s. Total resized so far:'
f' {total}')
- await asyncio.gather(*tasks)
consumer.commit_offsets()
