KeyedStateCheckpointOutputStream extends NonClosingCheckpointOutputStream<KeyGroupsStateHandle> { @Override KeyGroupsStateHandle closeAndGetHandle() throws IOException { StreamStateHandle streamStateHandle = delegate.closeAndGetHandle(); return streamStateHandle != null ? new KeyGroupsStateHandle(keyGroupRangeOffsets, streamStateHandle) : null; } KeyedStateCheckpointOutputStream(
			CheckpointStreamFactory.CheckpointStateOutputStream delegate, KeyGroupRange keyGroupRange); @Override void close(); KeyGroupsList getKeyGroupList(); void startNewKeyGroup(int keyGroupId); boolean isKeyGroupAlreadyStarted(int keyGroupId); boolean isKeyGroupAlreadyFinished(int keyGroupId); int getCurrentKeyGroup(); static final long NO_OFFSET_SET; static final int NO_CURRENT_KEY_GROUP; }
@Test public void testEmptyKeyedStream() throws Exception { final KeyGroupRange keyRange = new KeyGroupRange(0, 2); KeyedStateCheckpointOutputStream stream = createStream(keyRange); TestMemoryCheckpointOutputStream innerStream = (TestMemoryCheckpointOutputStream) stream.getDelegate(); KeyGroupsStateHandle emptyHandle = stream.closeAndGetHandle(); Assert.assertTrue(innerStream.isClosed()); Assert.assertEquals(null, emptyHandle); }