docs: @SneakyThrows
     public Corpus process(@NonNull Corpus documents) {
        timer.start();
  
        Broker.Builder<Document> builder = Broker.<Document>builder()
                                              .addProducer(new IterableProducer<>(documents))
                                              .bufferSize(queueSize);
  
        Corpus corpus = documents;
        if (returnCorpus && corpus.getDataSetType() == DatasetType.OffHeap) {
           Resource tempFile = Resources.temporaryDirectory();
           tempFile.deleteOnExit();
           try (MultiFileWriter writer = new MultiFileWriter(tempFile, "part-", Config.get("files.partition")
                                                                                      .asIntegerValue(numberOfThreads))) {
              builder.addConsumer(new AnnotateConsumer(annotationTypes, onComplete, documentsProcessed, wordsProcessed, writer),
                                  numberOfThreads)
                     .build()
                     .run();
           }
           corpus = Corpus.builder().offHeap().source(CorpusFormats.JSON_OPL, tempFile).build();
        } else {
           builder.addConsumer(new AnnotateConsumer(annotationTypes, onComplete, documentsProcessed,wordsProcessed, null),
                               numberOfThreads)
                  .build()
                  .run();
        }
  
        timer.stop();
        totalTime += timer.elapsed(TimeUnit.NANOSECONDS);
        timer.reset();
  
        return corpus;
     }