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());
        }
    }
@Test
    public void shouldThrowIOExceptionWhenWritingToNotExistedFile() {
        final Map<TopicPartition, Long> offsetsToWrite = Collections.singletonMap(new TopicPartition(topic, 0), 0L);

        final File notExistedFile = new File("/not_existed_dir/not_existed_file");
        final OffsetCheckpoint checkpoint = new OffsetCheckpoint(notExistedFile);
        
        final IOException e = assertThrows(IOException.class, () -> checkpoint.write(offsetsToWrite));
        assertThat(e.getMessage(), containsString("No such file or directory"));
    }