Out of memory on backfill from huge bucket

We acknowledge each buffer just before writing to Kafka, but also the records collected into batches. And when the poller cannot empty the queue of incoming DCP messages, it ends acknowledging without actually finishing batch for Kafka topic, which leads to OOM.