public PartitionMetadata read() {
        synchronized (lock) {
            try {
                try (BufferedReader reader = Files.newBufferedReader(path(), StandardCharsets.UTF_8)) {
                    PartitionMetadataReadBuffer partitionBuffer = new PartitionMetadataReadBuffer(file.getAbsolutePath(), reader);
                    return partitionBuffer.read();
                }
            } catch (IOException e) {
                String msg = "Error while reading partition metadata file " + file.getAbsolutePath();
                logDirFailureChannel.maybeAddOfflineLogDir(logDir(), msg, e);
                throw new KafkaStorageException(msg, e);
            }
        }
    }
@Test
    public void testRead() {
        LogDirFailureChannel channel = Mockito.mock(LogDirFailureChannel.class);
        PartitionMetadataFile partitionMetadataFile = new PartitionMetadataFile(file, channel);

        Uuid topicId = Uuid.randomUuid();
        partitionMetadataFile.record(topicId);
        partitionMetadataFile.maybeFlush();

        PartitionMetadata metadata = partitionMetadataFile.read();
        assertEquals(0, metadata.version());
        assertEquals(topicId, metadata.topicId());
    }