public void write(final Map<TopicPartition, Long> offsets) throws IOException {
        // if there are no offsets, skip writing the file to save disk IOs
        // but make sure to delete the existing file if one exists
        if (offsets.isEmpty()) {
            Utils.delete(file);
            return;
        }

        synchronized (lock) {
            // write to temp file and then swap with the existing file
            final File temp = new File(file.getAbsolutePath() + ".tmp");
            LOG.trace("Writing tmp checkpoint file {}", temp.getAbsolutePath());

            final FileOutputStream fileOutputStream = new FileOutputStream(temp);
            try (final BufferedWriter writer = new BufferedWriter(
                    new OutputStreamWriter(fileOutputStream, StandardCharsets.UTF_8))) {
                writeIntLine(writer, VERSION);
                writeIntLine(writer, offsets.size());

                for (final Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
                    final TopicPartition tp = entry.getKey();
                    final Long offset = entry.getValue();
                    if (isValid(offset)) {
                        writeEntry(writer, tp, offset);
                    } else {
                        LOG.error("Received offset={} to write to checkpoint file for {}", offset, tp);
                        throw new IllegalStateException("Attempted to write a negative offset to the checkpoint file");
                    }
                }

                writer.flush();
                fileOutputStream.getFD().sync();
            }

            LOG.trace("Swapping tmp checkpoint file {} {}", temp.toPath(), file.toPath());
            Utils.atomicMoveWithFallback(temp.toPath(), file.toPath());
        }
    }
Write the given offsets to the checkpoint file. All offsets should be non-negative.

@throws IOException if any file operation fails with an IO exception
streams/src/main/java/org/apache/kafka/streams/state/internals/OffsetCheckpoint.java