void publishMetricsFromQueue() {
        int retryDelay = firstRetryDelay;
        int n = retryCount;
        int minDelay = Math.min(500, retryDelay * 1000); // millis
        Random rng = new Random(System.nanoTime());
        while (!stopping) {
            try {
                queue.consumeAll(consumer);
                retryDelay = firstRetryDelay;
                n = retryCount;
                inError = false;
            } catch (InterruptedException e) {
                LOG.info(name + " thread interrupted.");
            } catch (Exception e) {
                if (n > 0) {
                    int awhile = rng.nextInt(retryDelay * 1000 - minDelay) + minDelay;
                    if (!inError) {
                        LOG.error("Got sink exception, retry in " + awhile + "ms", e);
                    }
                    retryDelay *= retryBackoff;
                    try {
                        Thread.sleep(awhile);
                    } catch (InterruptedException e2) {
                        LOG.info(name + " thread interrupted while waiting for retry", e2);
                    }
                    --n;
                } else {
                    if (!inError) {
                        LOG.error("Got sink exception and over retry limit, " + "suppressing further error messages", e);
                    }
                    queue.clear();
                    inError = true; // Don't keep complaining ad infinitum
                }
            } finally {
                qsize.set(queue.size());
            }
        }
    }