private synchronized NamedCache getOrCreateCache(final String name) {
        NamedCache cache = caches.get(name);
        if (cache == null) {
            cache = new NamedCache(name, this.metrics);
            caches.put(name, cache);
        }
        return cache;
    }
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