public LRUCacheEntry putIfAbsent(final String namespace, final Bytes key, final LRUCacheEntry value) {
        final NamedCache cache = getOrCreateCache(namespace);
        final LRUCacheEntry result;
        synchronized (cache) {
            final long oldSize = cache.sizeInBytes();
            result = cache.putIfAbsent(key, value);
            sizeInBytes.getAndAdd(cache.sizeInBytes() - oldSize);
            maybeEvict(namespace, cache);
        }

        if (result == null) {
            numPuts++;
        }
        return result;
    }
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