private void loadFromSnapshot(long logStartOffset, long currentTime) throws IOException {
        while (true) {
            Optional<SnapshotFile> latestSnapshotFileOptional = latestSnapshotFile();
            if (latestSnapshotFileOptional.isPresent()) {
                SnapshotFile snapshot = latestSnapshotFileOptional.get();
                try {
                    log.info("Loading producer state from snapshot file '{}'", snapshot);
                    Stream<ProducerStateEntry> loadedProducers = readSnapshot(snapshot.file()).stream().filter(producerEntry -> !isProducerExpired(currentTime, producerEntry));
                    loadedProducers.forEach(this::loadProducerEntry);
                    lastSnapOffset = snapshot.offset;
                    lastMapOffset = lastSnapOffset;
                    updateOldestTxnTimestamp();
                    return;
                } catch (CorruptSnapshotException e) {
                    log.warn("Failed to load producer snapshot from '{}': {}", snapshot.file(), e.getMessage());
                    removeAndDeleteSnapshot(snapshot.offset);
                }
            } else {
                lastSnapOffset = logStartOffset;
                lastMapOffset = logStartOffset;
                return;

            }
        }
    }
Get an unmodifiable map of active producers.
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java