@Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        // when we are restoring state with pendingCommitTransactions, we don't really know whether
        // the
        // transactions were already committed, or whether there was a failure between
        // completing the checkpoint on the master, and notifying the writer here.

        // (the common case is actually that is was already committed, the window
        // between the commit on the master and the notification here is very small)

        // it is possible to not have any transactions at all if there was a failure before
        // the first completed checkpoint, or in case of a scale-out event, where some of the
        // new task do not have and transactions assigned to check)

        // we can have more than one transaction to check in case of a scale-in event, or
        // for the reasons discussed in the 'notifyCheckpointComplete()' method.

        state = context.getOperatorStateStore().getListState(stateDescriptor);

        boolean recoveredUserContext = false;
        if (context.isRestored()) {
            LOG.info("{} - restoring state", name());
            for (State<TXN, CONTEXT> operatorState : state.get()) {
                userContext = operatorState.getContext();
                List<TransactionHolder<TXN>> recoveredTransactions =
                        operatorState.getPendingCommitTransactions();
                List<TXN> handledTransactions = new ArrayList<>(recoveredTransactions.size() + 1);
                for (TransactionHolder<TXN> recoveredTransaction : recoveredTransactions) {
                    // If this fails to succeed eventually, there is actually data loss
                    recoverAndCommitInternal(recoveredTransaction);
                    handledTransactions.add(recoveredTransaction.handle);
                    LOG.info("{} committed recovered transaction {}", name(), recoveredTransaction);
                }

                {
                    if (operatorState.getPendingTransaction() != null) {
                        TXN transaction = operatorState.getPendingTransaction().handle;
                        recoverAndAbort(transaction);
                        handledTransactions.add(transaction);
                        LOG.info(
                                "{} aborted recovered transaction {}",
                                name(),
                                operatorState.getPendingTransaction());
                    }
                }

                if (userContext.isPresent()) {
                    finishRecoveringContext(handledTransactions);
                    recoveredUserContext = true;
                }
            }
        }

        // if in restore we didn't get any userContext or we are initializing from scratch
        if (!recoveredUserContext) {
            LOG.info("{} - no state to restore", name());

            userContext = initializeUserContext();
        }
        this.pendingCommitTransactions.clear();

        currentTransactionHolder = beginTransactionInternal();
        LOG.debug("{} - started new transaction '{}'", name(), currentTransactionHolder);
    }
@Test
    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);

        assertThatThrownBy(
                        () -> {
                            harness.processElement("44", 4);
                            harness.snapshot(2, 5);
                        })
                .hasCauseInstanceOf(ContentDump.NotWritableException.class);

        closeTestHarness();

        tmpDirectory.setWritable(true);

        setUpTestHarness();
        harness.initializeState(snapshot);

        assertExactlyOnce(Arrays.asList("42", "43"));
        closeTestHarness();

        assertThat(tmpDirectory.listFiles()).isEmpty();
    }