### Question:
TwoPhaseCommitSinkFunction extends RichSinkFunction<IN> implements CheckpointedFunction, CheckpointListener { @Override public void initializeState(FunctionInitializationContext context) throws Exception { state = context.getOperatorStateStore().getListState(stateDescriptor); if (context.isRestored()) { LOG.info("{} - restoring state", name()); for (State<TXN, CONTEXT> operatorState : state.get()) { userContext = operatorState.getContext(); List<TransactionHolder<TXN>> recoveredTransactions = operatorState.getPendingCommitTransactions(); for (TransactionHolder<TXN> recoveredTransaction : recoveredTransactions) { recoverAndCommitInternal(recoveredTransaction); LOG.info("{} committed recovered transaction {}", name(), recoveredTransaction); } recoverAndAbort(operatorState.getPendingTransaction().handle); LOG.info("{} aborted recovered transaction {}", name(), operatorState.getPendingTransaction()); if (userContext.isPresent()) { finishRecoveringContext(); } } } if (userContext == null) { LOG.info("{} - no state to restore", name()); userContext = initializeUserContext(); } this.pendingCommitTransactions.clear(); currentTransactionHolder = beginTransactionInternal(); LOG.debug("{} - started new transaction '{}'", name(), currentTransactionHolder); } TwoPhaseCommitSinkFunction(
			TypeSerializer<TXN> transactionSerializer,
			TypeSerializer<CONTEXT> contextSerializer); @VisibleForTesting  TwoPhaseCommitSinkFunction(
		TypeSerializer<TXN> transactionSerializer,
		TypeSerializer<CONTEXT> contextSerializer,
		Clock clock); @Override final void invoke(IN value); @Override final void invoke(
		IN value, Context context); @Override final void notifyCheckpointComplete(long checkpointId); @Override void snapshotState(FunctionSnapshotContext context); @Override void initializeState(FunctionInitializationContext context); @Override void close();  }### Answer:
@Test public void testFailBeforeNotify() throws Exception { harness.open(); harness.processElement("42", 0); harness.snapshot(0, 1); harness.processElement("43", 2); OperatorSubtaskState snapshot = harness.snapshot(1, 3); tmpDirectory.setWritable(false); try { harness.processElement("44", 4); harness.snapshot(2, 5); fail("something should fail"); } catch (Exception ex) { if (!(ex.getCause() instanceof ContentDump.NotWritableException)) { throw ex; } } closeTestHarness(); tmpDirectory.setWritable(true); setUpTestHarness(); harness.initializeState(snapshot); assertExactlyOnce(Arrays.asList("42", "43")); closeTestHarness(); assertEquals(0, tmpDirectory.listFiles().size()); }