@Override
    KeyGroupsStateHandle closeAndGetHandle() throws IOException {
        StreamStateHandle streamStateHandle = super.closeAndGetHandleAfterLeasesReleased();
        return streamStateHandle != null
                ? new KeyGroupsStateHandle(keyGroupRangeOffsets, streamStateHandle)
                : null;
    }
@Test
    void testEmptyKeyedStream() throws Exception {
        final KeyGroupRange keyRange = new KeyGroupRange(0, 2);
        KeyedStateCheckpointOutputStream stream = createStream(keyRange);
        TestMemoryCheckpointOutputStream innerStream =
                (TestMemoryCheckpointOutputStream) stream.getDelegate();
        KeyGroupsStateHandle emptyHandle = stream.closeAndGetHandle();
        assertThat(innerStream.isClosed()).isTrue();
        assertThat(emptyHandle).isNull();
    }