class BlockBasedRamAccounting implements RamAccounting {

    public static final int MAX_BLOCK_SIZE_IN_BYTES = 2 * 1024 * 1024;

    /**
     * 2 GB × 0.001 ➞ MB
     *   = 2 MB
     *
     * 256 MB × 0.001 ➞ MB
     *
     *   = 0.256 MB
     */
    private static final double BREAKER_LIMIT_PERCENTAGE = 0.001;

    private final LongConsumer reserveMemory;
    private final int blockSizeInBytes;

    private long usedBytes = 0;
    private long reservedBytes = 0;

    /**
     * Scale the block size that is eagerly allocated depending on the circuit breakers current limit and the shards
     * on the node.
     */
    public static int blockSizeInBytesPerShard(long breakerLimit, int shardsUsedOnNode) {
        int blockSize = Math.min((int) (breakerLimit * BREAKER_LIMIT_PERCENTAGE), MAX_BLOCK_SIZE_IN_BYTES);
        return blockSize / Math.max(shardsUsedOnNode, 1);
    }

    public static int blockSizeInBytes(long breakerLimit) {
        return Math.min((int) (breakerLimit * BREAKER_LIMIT_PERCENTAGE), MAX_BLOCK_SIZE_IN_BYTES);
    }

    public BlockBasedRamAccounting(LongConsumer reserveMemory, int blockSizeInBytes) {
        this.reserveMemory = reserveMemory;
        this.blockSizeInBytes = blockSizeInBytes;
    }

    @Override
    public void addBytes(long bytes) {<FILL_FUNCTION_BODY>}

    @Override
    public long totalBytes() {
        return usedBytes;
    }

    @Override
    public void release() {
        reserveMemory.accept(- reservedBytes);
        usedBytes = 0;
        reservedBytes = 0;
    }

    @Override
    public void close() {
        release();
    }

    @Override
    public String toString() {
        return "RamAccounting{" +
               "usedBytes=" + usedBytes +
               ", reservedBytes=" + reservedBytes +
               '}';
    }
}

        usedBytes += bytes;
        if ((reservedBytes - usedBytes) < 0) {
            long reserveBytes = bytes > blockSizeInBytes ? bytes : blockSizeInBytes;
            try {
                reserveMemory.accept(reserveBytes);
            } catch (Exception e) {
                usedBytes -= bytes;
                throw e;
            }
            reservedBytes += reserveBytes;
        }
        assert reservedBytes >= usedBytes : "reservedBytes must be >= usedBytes: " + toString();
    
<no_super_class>