public Optional<File> takeSnapshot(boolean sync) throws IOException {
        // If not a new offset, then it is not worth taking another snapshot
        if (lastMapOffset > lastSnapOffset) {
            SnapshotFile snapshotFile = new SnapshotFile(LogFileUtils.producerSnapshotFile(logDir, lastMapOffset));
            long start = time.hiResClockMs();
            writeSnapshot(snapshotFile.file(), producers, sync);
            if (log.isDebugEnabled()) {
                log.debug("Wrote producer snapshot at offset {} with {} producer ids in {} ms.", lastMapOffset,
                    producers.size(), time.hiResClockMs() - start);
            }

            snapshots.put(snapshotFile.offset, snapshotFile);

            // Update the last snap offset according to the serialized map
            lastSnapOffset = lastMapOffset;

            return Optional.of(snapshotFile.file());
        }
        return Optional.empty();
    }
Take a snapshot at the current end offset if one does not already exist, then return the snapshot file if taken.
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java