@ProcessElement
  public void processElement(ProcessContext c) throws IOException, InterruptedException {
    Instant firstInstant = c.sideInput(firstInstantSideInput);

    T element = c.element();
    BackOff backoff = fluentBackoff.backoff();
    while (true) {
      Instant instant = Instant.now();
      int maxOpsBudget = calcMaxOpsBudget(firstInstant, instant, this.numWorkers.get());
      long currentOpCount = successfulOps.get(instant.getMillis());
      long availableOps = maxOpsBudget - currentOpCount;

      if (maxOpsBudget >= Integer.MAX_VALUE || availableOps > 0) {
        c.output(element);
        successfulOps.add(instant.getMillis(), 1);
        return;
      }

      long backoffMillis = backoff.nextBackOffMillis();
      LOG.info("Delaying by {}ms to conform to gradual ramp-up.", backoffMillis);
      throttlingMsecs.inc(backoffMillis);
      sleeper.sleep(backoffMillis);
    }
  }
@Test
  public void testRampupThrottler() throws Exception {
    Map<Duration, Integer> rampupSchedule =
        ImmutableMap.<Duration, Integer>builder()
            .put(Duration.ZERO, 500)
            .put(Duration.millis(1), 0)
            .put(Duration.standardSeconds(1), 500)
            .put(Duration.standardSeconds(1).plus(Duration.millis(1)), 0)
            .put(Duration.standardMinutes(5), 500)
            .put(Duration.standardMinutes(10), 750)
            .put(Duration.standardMinutes(15), 1125)
            .put(Duration.standardMinutes(30), 3796)
            .put(Duration.standardMinutes(60), 43248)
            .build();

    for (Map.Entry<Duration, Integer> entry : rampupSchedule.entrySet()) {
      DateTimeUtils.setCurrentMillisFixed(entry.getKey().getMillis());
      for (int i = 0; i < entry.getValue(); i++) {
        rampupThrottlingFnTester.processElement(UUID.randomUUID().toString());
      }
      assertThrows(RampupDelayException.class, () -> rampupThrottlingFnTester.processElement(null));
    }
  }