double nextRate(double rate, long periodNanos, long totalPublished, long totalReceived) {
        long expected = (long) ((rate / ONE_SECOND_IN_NANOS) * periodNanos);
        long published = totalPublished - previousTotalPublished;
        long received = totalReceived - previousTotalReceived;

        previousTotalPublished = totalPublished;
        previousTotalReceived = totalReceived;

        if (log.isDebugEnabled()) {
            log.debug(
                    "Current rate: {} -- Publish rate {} -- Receive Rate: {}",
                    rate,
                    rate(published, periodNanos),
                    rate(received, periodNanos));
        }

        long receiveBacklog = totalPublished - totalReceived;
        if (receiveBacklog > receiveBacklogLimit) {
            return nextRate(periodNanos, received, expected, receiveBacklog, "Receive");
        }

        long publishBacklog = expected - published;
        if (publishBacklog > publishBacklogLimit) {
            return nextRate(periodNanos, published, expected, publishBacklog, "Publish");
        }

        rampUp();

        return rate + (rate * rampingFactor);
    }
@Test
    void receiveBacklog() {
        assertThat(rateController.getRampingFactor()).isEqualTo(1);

        // no backlog
        rate = rateController.nextRate(rate, periodNanos, 10_000, 10_000);
        assertThat(rate).isEqualTo(20_000);
        assertThat(rateController.getRampingFactor()).isEqualTo(1);

        // receive backlog
        rate = rateController.nextRate(rate, periodNanos, 20_000, 15_000);
        assertThat(rate).isEqualTo(5_000);
        assertThat(rateController.getRampingFactor()).isEqualTo(0.5);
    }