public void truncateFullyAndStartAt(long offset) throws IOException {
        clearProducerIds();
        ongoingTxns.clear();
        unreplicatedTxns.clear();
        for (SnapshotFile snapshotFile : snapshots.values()) {
            removeAndDeleteSnapshot(snapshotFile.offset);
        }
        lastSnapOffset = 0L;
        lastMapOffset = offset;
        updateOldestTxnTimestamp();
    }
@Test
    public void testTruncateFullyAndStartAt() throws IOException {
        appendClientEntry(stateManager, producerId, epoch, defaultSequence, 0L, false);
        appendClientEntry(stateManager, producerId, epoch, 1, 1L, false);
        stateManager.takeSnapshot();
        assertEquals(1, Objects.requireNonNull(logDir.listFiles()).length);
        assertEquals(singleton(2L), currentSnapshotOffsets());

        appendClientEntry(stateManager, producerId, epoch, 2, 2L, false);
        stateManager.takeSnapshot();
        assertEquals(2, Objects.requireNonNull(logDir.listFiles()).length);
        assertEquals(new HashSet<>(asList(2L, 3L)), currentSnapshotOffsets());

        stateManager.truncateFullyAndStartAt(0L);

        assertEquals(0, Objects.requireNonNull(logDir.listFiles()).length);
        assertEquals(emptySet(), currentSnapshotOffsets());

        appendClientEntry(stateManager, producerId, epoch, 0, 0L, false);
        stateManager.takeSnapshot();
        assertEquals(1, Objects.requireNonNull(logDir.listFiles()).length);
        assertEquals(singleton(1L), currentSnapshotOffsets());
    }