can't you make StateTransferManagerImpl.applyInMemoryState() to always write to the cache store and modify StateTransferManagerImpl.generatePersistentState to skip sending a key if it is already in the memory/DataContainer?
That would require passing a filter to CacheStore.toStrem…