public void maybeFlush() {
        // We check dirtyTopicId first to avoid having to take the lock unnecessarily in the frequently called log append path
        if (dirtyTopicIdOpt.isPresent()) {
            // We synchronize on the actual write to disk
            synchronized (lock) {
                dirtyTopicIdOpt.ifPresent(topicId -> {
                    try {
                        try (FileOutputStream fileOutputStream = new FileOutputStream(tempPath().toFile());
                             BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(fileOutputStream, StandardCharsets.UTF_8))) {
                            writer.write(new PartitionMetadata(CURRENT_VERSION, topicId).encode());
                            writer.flush();
                            fileOutputStream.getFD().sync();
                        }

                        Utils.atomicMoveWithFallback(tempPath(), path());
                    } catch (IOException e) {
                        String msg = "Error while writing partition metadata file " + file.getAbsolutePath();
                        logDirFailureChannel.maybeAddOfflineLogDir(logDir(), msg, e);
                        throw new KafkaStorageException(msg, e);
                    }
                    dirtyTopicIdOpt = Optional.empty();
                });
            }
        }
    }
@Test
    public void testMaybeFlushWithNoTopicIdPresent() {
        PartitionMetadataFile partitionMetadataFile = new PartitionMetadataFile(file, null);
        partitionMetadataFile.maybeFlush();

        assertEquals(0, file.length());
    }