code: private CompletableFuture<ProcessorCollection> getProcessor(long streamSegmentId) {
          ProcessorCollection existingProcessor = this.processors.getOrDefault(streamSegmentId, null);
          if (existingProcessor != null) {
              if (closeIfNecessary(existingProcessor).isClosed()) {
                  // Existing SegmentAggregator has become stale (most likely due to its SegmentMetadata being evicted),
                  // so it has been closed and we need to create a new one.
                  this.processors.remove(streamSegmentId);
              } else {
                  return CompletableFuture.completedFuture(existingProcessor);
              }
          }
  
          // Get the SegmentAggregator's Metadata.
          UpdateableSegmentMetadata segmentMetadata = this.dataSource.getStreamSegmentMetadata(streamSegmentId);
          if (segmentMetadata == null) {
              return Futures.failedFuture(new DataCorruptionException(String.format(
                      "No StreamSegment with id '%d' is registered in the metadata.", streamSegmentId)));
          }
  
          // Then create the aggregator, and only register it after a successful initialization. Otherwise we risk
          // having a registered aggregator that is not initialized.
          SegmentAggregator newAggregator = new SegmentAggregator(segmentMetadata, this.dataSource, this.storage, this.config, this.timer, this.executor);
          ProcessorCollection pc = new ProcessorCollection(newAggregator, this.createProcessors.apply(segmentMetadata));
          try {
              CompletableFuture<Void> init = newAggregator.initialize(this.config.getFlushTimeout());
              Futures.exceptionListener(init, ex -> newAggregator.close());
              return init.thenApply(ignored -> {
                  this.processors.put(streamSegmentId, pc);
                  return pc;
              });
          } catch (Exception ex) {
              pc.close();
              throw ex;
          }
      }
docs: Gets, or creates, a SegmentAggregator for the given StorageOperation.
  
  @param streamSegmentId The Id of the StreamSegment to get the aggregator for.