private void maybeEvict(final String namespace, final NamedCache cache) {
        int numEvicted = 0;
        while (sizeInBytes.get() > maxCacheSizeBytes) {
            // we abort here as the put on this cache may have triggered
            // a put on another cache. So even though the sizeInBytes() is
            // still > maxCacheSizeBytes there is nothing to evict from this
            // namespaced cache.
            if (cache.isEmpty()) {
                return;
            }
            final long oldSize = cache.sizeInBytes();
            cache.evict();
            sizeInBytes.getAndAdd(cache.sizeInBytes() - oldSize);
            numEvicts++;
            numEvicted++;
        }
        if (log.isTraceEnabled()) {
            log.trace("Evicted {} entries from cache {}", numEvicted, namespace);
        }
    }
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