code: private synchronized CompletableFuture<Void> compact(Compaction compaction, CompletableFuture<Void> future, ThreadContext context) {
      CompactionManager manager = compaction.manager(this);
      AtomicInteger counter = new AtomicInteger();
  
      Collection<CompactionTask> tasks = manager.buildTasks(storage, segments);
      if (!tasks.isEmpty()) {
        LOGGER.info("Compacting log with compaction: {}", compaction);
        LOGGER.debug("Executing {} compaction task(s)", tasks.size());
        for (CompactionTask task : tasks) {
          LOGGER.debug("Executing {}", task);
          ThreadContext taskThread = new ThreadPoolContext(executor, segments.serializer());
          taskThread.execute(task).whenComplete((result, error) -> {
            LOGGER.debug("{} complete", task);
            if (counter.incrementAndGet() == tasks.size()) {
              if (context != null) {
                context.executor().execute(() -> future.complete(null));
              } else {
                future.complete(null);
              }
            }
          });
        }
      } else {
        future.complete(null);
      }
      return future;
    }
docs: Compacts the log.