@Override
    public final void finish() throws Exception {
        finished = true;
        finishProcessing(currentTransaction());
    }
@Test
    void testNoTransactionAfterSinkFunctionFinish() throws Exception {
        harness.open();
        harness.processElement("42", 0);
        harness.snapshot(0, 1);
        harness.processElement("43", 2);
        harness.snapshot(1, 3);
        harness.processElement("44", 4);

        // do not expect new input after finish()
        sinkFunction.finish();

        harness.snapshot(2, 5);
        harness.notifyOfCompletedCheckpoint(1);

        // make sure the previous empty transaction will not be pre-committed
        harness.snapshot(3, 6);

        assertThatThrownBy(() -> harness.processElement("45", 7))
                .isInstanceOf(NullPointerException.class);

        // Checkpoint2 has not complete
        assertExactlyOnce(Arrays.asList("42", "43"));

        // transaction for checkpoint2
        assertThat(tmpDirectory.listFiles()).hasSize(1);
    }