code: private void cleanup() {
          long traceId = LoggerHelpers.traceEnterWithContext(log, this.traceObjectId, "cleanup");
          val toRemove = this.processors.values().stream()
                                        .map(this::closeIfNecessary)
                                        .filter(ProcessorCollection::isClosed)
                                        .map(ProcessorCollection::getId)
                                        .collect(Collectors.toList());
          toRemove.forEach(this.processors::remove);
          LoggerHelpers.traceLeave(log, this.traceObjectId, "cleanup", traceId, toRemove.size());
      }
docs: Cleans up all SegmentAggregators that are currently closed.