@Override
    public void collectSizeStats(StateObjectSizeStatsCollector collector) {
        streamSubHandles().forEach(handle -> handle.collectSizeStats(collector));
    }
@Test
    void testCollectSizeStats() {
        IncrementalRemoteKeyedStateHandle handle = create(ThreadLocalRandom.current());
        StateObject.StateObjectSizeStatsCollector statsCollector =
                StateObject.StateObjectSizeStatsCollector.create();
        handle.collectSizeStats(statsCollector);
        Assertions.assertEquals(
                new HashMap<StateObject.StateObjectLocation, Long>() {
                    {
                        // Location is LOCAL_MEMORY, even though the handle is called remote because
                        // we test against a local file system
                        put(StateObject.StateObjectLocation.LOCAL_MEMORY, handle.getStateSize());
                    }
                },
                statsCollector.getStats());
    }