@Override
    public long throttledByThroughput(final long bytes) {
        long availableSize;
        final long nowUs = Utils.monotonicUs();
        final long limitPerCycle = this.throttleThroughputBytes / this.checkCycleSecs;
        this.lock.lock();
        try {
            if (this.currThroughputBytes + bytes > limitPerCycle) {
                // reading another |bytes| exceeds the limit
                if (nowUs - this.lastThroughputCheckTimeUs <= 1000 * 1000 / this.checkCycleSecs) {
                    // if time interval is less than or equal to a cycle, read more data
                    // to make full use of the throughput of current cycle.
                    availableSize = limitPerCycle - this.currThroughputBytes;
                    this.currThroughputBytes = limitPerCycle;
                } else {
                    // otherwise, read the data in the next cycle.
                    availableSize = bytes > limitPerCycle ? limitPerCycle : bytes;
                    this.currThroughputBytes = availableSize;
                    this.lastThroughputCheckTimeUs = calculateCheckTimeUs(nowUs);
                }
            } else {
                // reading another |bytes| doesn't exceed limit(less than or equal to),
                // put it in current cycle
                availableSize = bytes;
                this.currThroughputBytes += availableSize;
            }
        } finally {
            this.lock.unlock();
        }
        return availableSize;
    }
@Test
    public void testThrottledByThroughput() throws Exception {
        assertEquals(512L, this.snapshotThrottle.throttledByThroughput(512));
        assertEquals(512L, this.snapshotThrottle.throttledByThroughput(512));
        assertEquals(0, this.snapshotThrottle.throttledByThroughput(512));
        Thread.sleep(1100);
        assertEquals(512L, this.snapshotThrottle.throttledByThroughput(512));
        assertEquals(512L, this.snapshotThrottle.throttledByThroughput(512));
        assertEquals(0, this.snapshotThrottle.throttledByThroughput(512));
    }