@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);
}