class ConcurrentRamAccounting implements RamAccounting {

    private final AtomicLong usedBytes = new AtomicLong(0L);
    private final LongConsumer reserveBytes;
    private final LongConsumer releaseBytes;
    private final String label;
    private final int operationMemoryLimit;

    public static ConcurrentRamAccounting forCircuitBreaker(String label, CircuitBreaker circuitBreaker, int operationMemoryLimit) {<FILL_FUNCTION_BODY>}

    public ConcurrentRamAccounting(LongConsumer reserveBytes,
                                   LongConsumer releaseBytes,
                                   String label,
                                   int operationMemoryLimit) {
        this.reserveBytes = reserveBytes;
        this.releaseBytes = releaseBytes;
        this.label = label;
        this.operationMemoryLimit = operationMemoryLimit;
    }

    @Override
    public void addBytes(long bytes) {
        long currentUsedBytes = usedBytes.addAndGet(bytes);
        if (operationMemoryLimit > 0 && currentUsedBytes > operationMemoryLimit) {
            usedBytes.addAndGet(- bytes);
            throw new CircuitBreakingException(String.format(Locale.ENGLISH,
                "\"%s\" reached operation memory limit. Used: %s, Limit: %s",
                label,
                new ByteSizeValue(currentUsedBytes),
                new ByteSizeValue(operationMemoryLimit)
            ));
        }
        try {
            reserveBytes.accept(bytes);
        } catch (Exception e) {
            usedBytes.addAndGet(- bytes);
            throw e;
        }
    }

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

    @Override
    public void release() {
        long prevUsedBytes = usedBytes.getAndSet(0);
        releaseBytes.accept(prevUsedBytes);
    }

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

        return new ConcurrentRamAccounting(
            bytes -> circuitBreaker.addEstimateBytesAndMaybeBreak(bytes, label),
            bytes -> circuitBreaker.addWithoutBreaking(- bytes),
            label,
            operationMemoryLimit
        );
    
<no_super_class>