@Test
public void testFailBeforeNotify() throws Exception {

	harness.open();
	harness.processElement("42", 0);
	harness.snapshot(0, 1);
	harness.processElement("43", 2);
	OperatorStateHandles snapshot = harness.snapshot(1, 3);
	assertTrue(tmpDirectory.setWritable(false));
	try {
		harness.processElement("44", 4);
		harness.snapshot(2, 5);
		fail("something should fail");
	}
	catch (Exception ex) {
		if (!(ex.getCause() instanceof FileNotFoundException)) {
			throw ex;
		}
		// ignore
	}
	closeTestHarness();
	assertTrue(tmpDirectory.setWritable(true));
	setUpTestHarness();
	harness.initializeState(snapshot);
	assertExactlyOnce(Arrays.asList("42", "43"));
	closeTestHarness();
	assertEquals(0, tmpDirectory.listFiles().length);
}