code: private CompletableFuture<TableWriterFlushResult> flushOnce(DirectSegmentAccess segment, TimeoutTimer timer) {
          // Index all the keys in the segment range pointed to by the aggregator.
          KeyUpdateCollection keyUpdates = readKeysFromSegment(segment, this.aggregator.getFirstOffset(), this.aggregator.getLastOffset(), timer);
          log.debug("{}: Flush.ReadFromSegment {} UpdateKeys(s).", this.traceObjectId, keyUpdates.getUpdates().size());
  
          // Group keys by their assigned TableBucket (whether existing or not), then fetch all existing keys
          // for each such bucket and finally (reindex) update the bucket.
          return this.indexWriter
                  .groupByBucket(segment, keyUpdates.getUpdates(), timer)
                  .thenComposeAsync(builders -> fetchExistingKeys(builders, segment, timer)
                                  .thenComposeAsync(v -> {
                                      val bucketUpdates = builders.stream().map(BucketUpdate.Builder::build).collect(Collectors.toList());
                                      logBucketUpdates(bucketUpdates);
                                      return this.indexWriter.updateBuckets(segment, bucketUpdates,
                                              this.aggregator.getLastIndexedOffset(), keyUpdates.getLastIndexedOffset(),
                                              keyUpdates.getTotalUpdateCount(), timer.getRemaining());
                                  }, this.executor),
                          this.executor)
                  .thenApply(ignored -> new TableWriterFlushResult(keyUpdates.getLastIndexedOffset(), keyUpdates.getHighestCopiedOffset()));
      }
docs: Performs a single flush attempt.
  
  @param segment A {@link DirectSegmentAccess} representing the Segment to flush on.
  @param timer   Timer for the operation.
  @return A CompletableFuture that, when completed, will indicate the flush has completed successfully. If the
  operation failed, it will be failed with the appropriate exception. Notable exceptions:
  <ul>
  <li>{@link BadAttributeUpdateException} If a conditional update on the {@link TableAttributes#INDEX_OFFSET} attribute failed.
  </ul>