created an aggregator that drops its aggregated message bodies after the producer of the route generated the last record