public void truncateAndReload(long logStartOffset, long logEndOffset, long currentTimeMs) throws IOException {
        // remove all out of range snapshots
        for (SnapshotFile snapshot : snapshots.values()) {
            if (snapshot.offset > logEndOffset || snapshot.offset <= logStartOffset) {
                removeAndDeleteSnapshot(snapshot.offset);
            }
        }

        if (logEndOffset != mapEndOffset()) {
            clearProducerIds();
            ongoingTxns.clear();
            updateOldestTxnTimestamp();

            // since we assume that the offset is less than or equal to the high watermark, it is
            // safe to clear the unreplicated transactions
            unreplicatedTxns.clear();
            loadFromSnapshot(logStartOffset, currentTimeMs);
        } else {
            onLogStartOffsetIncremented(logStartOffset);
        }
    }
Truncate the producer id mapping to the given offset range and reload the entries from the most recent
snapshot in range (if there is one). We delete snapshot files prior to the logStartOffset but do not remove
producer state from the map. This means that in-memory and on-disk state can diverge, and in the case of
broker failover or unclean shutdown, any in-memory state not persisted in the snapshots will be lost, which
would lead to UNKNOWN_PRODUCER_ID errors. Note that the log end offset is assumed to be less than or equal
to the high watermark.
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java