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;
    }
public FileWriter getOrCreateWriter ( LogFilePath path , CompressionCodec codec ) throws Exception { FileWriter writer = mWriters . get ( path ) ; if ( writer == null ) { 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 ; }
Retrieve a writer for a given path or create a new one if it does not exist.
@param path The path to retrieve writer for.
@param codec Optional compression codec.
@return Writer for a given path.
@throws Exception on error