public void flush(final String namespace) {
        numFlushes++;

        final NamedCache cache = getCache(namespace);
        if (cache == null) {
            return;
        }

        synchronized (cache) {
            final long oldSize = cache.sizeInBytes();
            cache.flush();
            sizeInBytes.getAndAdd(cache.sizeInBytes() - oldSize);
        }

        if (log.isTraceEnabled()) {
            log.trace("Cache stats on flush: #puts={}, #gets={}, #evicts={}, #flushes={}", puts(), gets(), evicts(), flushes());
        }
    }
Add a listener that is called each time an entry is evicted from the cache or an explicit flush is called
streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java