@Override
    public CompletableFuture<SnapshotResult<ChangelogStateHandleStreamImpl>> persist(
            SequenceNumber from, long checkpointId) throws IOException {
        LOG.debug(
                "persist {} starting from sqn {} (incl.), active sqn: {}",
                logId,
                from,
                activeSequenceNumber);
        return persistInternal(from, checkpointId);
    }
@Test
    void testPersist() throws Exception {
        withWriter(
                (writer, uploader) -> {
                    byte[] bytes = getBytes();
                    CompletableFuture<SnapshotResult<ChangelogStateHandleStreamImpl>> future =
                            writer.persist(append(writer, bytes), 1L);
                    assertSubmittedOnly(uploader, bytes);
                    uploader.completeUpload();
                    assertThat(
                                    getOnlyElement(
                                                    future.get()
                                                            .getJobManagerOwnedSnapshot()
                                                            .getHandlesAndOffsets())
                                            .f0
                                            .asBytesIfInMemory()
                                            .get())
                            .isEqualTo(bytes);
                });
    }