public void write(final ConsumerRecord<byte[], byte[]> record) throws IOException {
    if (!writable) {
      throw new IOException("Write permission denied.");
    }
    final File dirty = dirty(file);
    final File tmp = tmp(file);
    // first write to the dirty copy
    appendRecordToFile(record, dirty, filesystem);
    // atomically rename the dirty copy to the "live" copy while copying the live copy to
    // the "dirty" copy via a temporary hard link
    Files.createLink(tmp.toPath(), file.toPath());
    Files.move(
        dirty.toPath(),
        file.toPath(),
        StandardCopyOption.REPLACE_EXISTING,
        StandardCopyOption.ATOMIC_MOVE
    );
    Files.move(tmp.toPath(), dirty.toPath());
    // keep the dirty copy in sync with the live copy, which now has the write
    appendRecordToFile(record, dirty, filesystem);
  }
@Test
  public void shouldWriteRecord() throws IOException {
    // Given
    final ConsumerRecord<byte[], byte[]> record = newStreamRecord("stream1");

    // When
    replayFile.write(record);

    // Then
    final List<String> commands = Files.readAllLines(internalReplayFile.toPath());
    assertThat(commands.size(), is(1));
    assertThat(commands.get(0), is(
        "\"stream/stream1/create\"" + KEY_VALUE_SEPARATOR
            + "{\"statement\":\"CREATE STREAM stream1 (id INT) WITH (kafka_topic='stream1')\""
            + ",\"streamsProperties\":{},\"originalProperties\":{},\"plan\":null}"
    ));
  }