@Override
    public long computeLocalQuota(long confUsage, long myUsage, long[] allUsages) throws PulsarAdminException {
        // ToDo: work out the initial conditions: we may allow a small number of "first few iterations" to go
        // unchecked as we get some history of usage, or follow some other "TBD" method.

        if (confUsage < 0) {
            // This can happen if the RG is not configured with this particular limit (message or byte count) yet.
            val retVal = -1;
            if (log.isDebugEnabled()) {
                log.debug("Configured usage ({}) is not set; returning a special value ({}) for calculated quota",
                        confUsage, retVal);
            }
            return retVal;
        }

        long totalUsage = 0;
        for (long usage : allUsages) {
            totalUsage += usage;
        }

        if (myUsage < 0 || totalUsage < 0) {
            String errMesg = String.format("Local usage (%d) or total usage (%d) is negative",
                    myUsage, totalUsage);
            log.error(errMesg);
            throw new PulsarAdminException(errMesg);
        }

        // If the total usage is zero (which may happen during initial transients), just return the configured value.
        // The caller is expected to check the value returned, or not call here with a zero global usage.
        // [This avoids a division by zero when calculating the local share.]
        if (totalUsage == 0) {
            if (log.isDebugEnabled()) {
                log.debug("computeLocalQuota: totalUsage is zero; "
                        + "returning the configured usage ({}) as new local quota",
                        confUsage);
            }
            return confUsage;
        }

        if (myUsage > totalUsage) {
            String errMesg = String.format("Local usage (%d) is greater than total usage (%d)",
                    myUsage, totalUsage);
            // Log as a warning [in case this can happen transiently (?)].
            log.warn(errMesg);
        }

        // How much unused capacity is left over?
        float residual = confUsage - totalUsage;

        // New quota is the old usage incremented by any residual as a ratio of the local usage to the total usage.
        // This should result in the calculatedQuota increasing proportionately if total usage is less than the
        // configured usage, and reducing proportionately if the total usage is greater than the configured usage.
        // Capped to 1, to prevent negative or zero setting of quota.
        // the rate limiter code assumes that rate value of 0 or less to mean that no rate limit should be applied
        float myUsageFraction = (float) myUsage / totalUsage;
        float calculatedQuota = max(myUsage + residual * myUsageFraction, 1);

        val longCalculatedQuota = (long) calculatedQuota;
        if (log.isDebugEnabled()) {
            log.debug("computeLocalQuota: myUsage={}, totalUsage={}, myFraction={}; newQuota returned={} [long: {}]",
                    myUsage, totalUsage, myUsageFraction, calculatedQuota, longCalculatedQuota);
        }

        return longCalculatedQuota;
    }
@Test
    public void testRQCalcNegativeConfTest() throws PulsarAdminException {
        final long[] allUsage = { 0 };
        long calculatedQuota = this.rqCalc.computeLocalQuota(-1, 0, allUsage);
        long expectedQuota = -1;
        Assert.assertEquals(calculatedQuota, expectedQuota);
    }