docs: public FileWriter getOrCreateWriter(LogFilePath path, CompressionCodec codec)
              throws Exception {
          FileWriter writer = mWriters.get(path);
          if (writer == null) {
              // Just in case.
              FileUtil.delete(path.getLogFilePath());
              FileUtil.delete(path.getLogFileCrcPath());
              TopicPartitionGroup topicPartition = new TopicPartitionGroup(path.getTopic(),
                      path.getKafkaPartitions());
              HashSet<LogFilePath> files = mFiles.get(topicPartition);
              if (files == null) {
                  files = new HashSet<LogFilePath>();
                  mFiles.put(topicPartition, files);
              }
              if (!files.contains(path)) {
                  files.add(path);
              }
              writer = ReflectionUtil.createFileWriter(mConfig.getFileReaderWriterFactory(), path, codec, mConfig);
              mWriters.put(path, writer);
              mCreationTimes.put(path, System.currentTimeMillis() / 1000L);
              LOG.debug("created writer for path {}", path.getLogFilePath());
              LOG.debug("Register deleteOnExit for path {}", path.getLogFilePath());
              FileUtil.deleteOnExit(path.getLogFileParentDir());
              FileUtil.deleteOnExit(path.getLogFileDir());
              FileUtil.deleteOnExit(path.getLogFilePath());
              FileUtil.deleteOnExit(path.getLogFileCrcPath());
          }
          return writer;
      }