@Override
    public FsCompletedCheckpointStorageLocation closeAndFinalizeCheckpoint() throws IOException {
        synchronized (this) {
            if (!closed) {
                try {
                    // make a best effort attempt to figure out the size
                    long size = 0;
                    try {
                        size = outputStreamWrapper.getOutput().getPos();
                    } catch (Exception ignored) {
                    }

                    outputStreamWrapper.closeForCommit();

                    FileStateHandle metaDataHandle = new FileStateHandle(metadataFilePath, size);

                    return new FsCompletedCheckpointStorageLocation(
                            fileSystem,
                            exclusiveCheckpointDir,
                            metaDataHandle,
                            metaDataHandle.getFilePath().getParent().toString());
                } catch (Exception e) {
                    try {
                        outputStreamWrapper.cleanup();
                    } catch (Exception deleteException) {
                        LOG.warn(
                                "Could not delete the checkpoint stream file {}.",
                                metadataFilePath,
                                deleteException);
                    }

                    throw new IOException(
                            "Could not flush and close the file system "
                                    + "output stream to "
                                    + metadataFilePath
                                    + " in order to obtain the "
                                    + "stream state handle",
                            e);
                } finally {
                    closed = true;
                }
            } else {
                throw new IOException("Stream has already been closed and discarded.");
            }
        }
    }
@TestTemplate
    void testFileExistence() throws Exception {
        Path metaDataFilePath = baseFolder();
        FsCheckpointMetadataOutputStream stream = createTestStream(metaDataFilePath, fileSystem);

        if (fileSystem instanceof FsWithoutRecoverableWriter) {
            assertThat(fileSystem.exists(metaDataFilePath)).isTrue();
        } else {
            assertThat(fileSystem.exists(metaDataFilePath)).isFalse();
        }

        stream.closeAndFinalizeCheckpoint();

        assertThat(fileSystem.exists(metaDataFilePath)).isTrue();
    }