@Override
  public double apply(double availableDiskQuotaPercentage) {
    double workFactor = entryIndex == -1 ? 1.0 : this.entries.get(entryIndex).getWorkFactor();
    int entryIndex = this.entryIndex;

    while (entryIndex >= 0 &&
        entries.get(entryIndex).getHighWaterMarkPercent() <= availableDiskQuotaPercentage) {
      --entryIndex;
    }

    while (entryIndex < entries.size() - 1 &&
        entries.get(entryIndex + 1).getLowWaterMarkPercent() > availableDiskQuotaPercentage) {
      ++entryIndex;
    }

    if (entryIndex != this.entryIndex) {
      workFactor = entryIndex == -1 ? 1.0 : entries.get(entryIndex).getWorkFactor();
      this.entryIndex = entryIndex;
      log.info("Work factor has been updated: {}.", workFactor);
    }

    return workFactor;
  }
Constructs a new watermark disk quota policy using the supplied policy entries.

@param entries a list of {@link Entry} objects that describe how to adjust the work rate. May
                be empty, but cannot be {@code null}.
samza-core/src/main/java/org/apache/samza/container/disk/WatermarkDiskQuotaPolicy.java