The short circuiting renormalizer used with ML anomaly detection will skip renormalizations if a more recent set of quantiles has been received.  It does this by maintaining a queue of quantiles documents, but when it is ready to perform the next renormalization it drains the entire queue so that it is left with the most recent.  When quantiles documents are large this is an inefficient use of memory.  Instead of storing the full quantiles documents in a queue it should just retain the most recently added quantiles document.
x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/AutodetectResultProcessor.java
throw new TimeoutException("Timed out waiting for results processor to complete for job " + jobId);
            }
            
            
            updateModelSnapshotSemaphore.acquire();
            updateModelSnapshotSemaphore.release();
            
            waitUntilRenormalizerIsIdle();
            persister.commitResultWrites(jobId);
            persister.commitAnnotationWrites();
            persister.commitStateWrites(jobId);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LOGGER.info("[{}] Interrupted waiting for results processor to complete", jobId);
        }
    }
    /**
     * Blocks until a flush is acknowledged or the timeout expires, whichever happens first.
     *
     * @param flushId the id of the flush request to wait for
     * @param timeout the timeout
     * @return The {@link FlushAcknowledgement} if the flush has completed or the parsing finished; {@code null} if the timeout expired
     */
    @Nullable
    public FlushAcknowledgement waitForFlushAcknowledgement(String flushId, Duration timeout) throws Exception {
        return failed ? null : flushListener.waitForFlush(flushId, timeout);
    }
    public void clearAwaitingFlush(String flushId) {
        flushListener.clear(flushId);
    }