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 ; }
Retrieve a writer for a given path or create a new one if it does not exist .