Merge pull request #70 from RADAR-base/release-1.0.1

Release 1.0.1

Author: nivemaham

build.gradle

@@ -9,7 +9,7 @@ plugins {
 }
 
 group 'org.radarbase'
-version '1.0.0'
+version '1.0.1'
 mainClassName = 'org.radarbase.output.Application'
 
 sourceCompatibility = '1.8'

src/main/java/org/radarbase/output/Application.kt

@@ -63,9 +63,7 @@ class Application(
     override val sourceStorage: SourceStorage
         get() = sourceStorageFactory.createSourceStorage()
 
-    private val targetStorageFactory = TargetStorageFactory(config.target)
-    override val targetStorage: TargetStorage
-            get() = targetStorageFactory.createTargetStorage()
+    override val targetStorage: TargetStorage = TargetStorageFactory(config.target).createTargetStorage()
 
     override val redisPool: JedisPool = JedisPool(config.redis.uri)
     override val remoteLockManager: RemoteLockManager = RedisRemoteLockManager(

src/main/java/org/radarbase/output/accounting/OffsetIntervals.kt

@@ -21,6 +21,7 @@ class OffsetIntervals {
     }
 
     fun contains(range: OffsetRangeSet.Range): Boolean {
+        checkNotNull(range.to)
         //  -index-1 if not found
         val searchIndex = offsetsFrom.binarySearch(range.from)
         val index = if (searchIndex >= 0) searchIndex else -searchIndex - 2
@@ -80,6 +81,7 @@ class OffsetIntervals {
 
     fun add(range: OffsetRangeSet.Range) {
         val (from, to, lastModified) = range
+        checkNotNull(to)
         var index = offsetsFrom.binarySearch(from)
         if (index < 0) {
             // index where this range would be entered

src/main/java/org/radarbase/output/accounting/OffsetRangeSet.kt

@@ -148,8 +148,8 @@ class OffsetRangeSet {
             .getOrDefault(this, EMPTY_VALUE)
             .read(function)
 
-    data class Range(val from: Long, val to: Long, val lastProcessed: Instant) {
-        val size: Long = to - from + 1
+    data class Range(val from: Long, val to: Long?, val lastProcessed: Instant) {
+        val size: Long? = to?.let { it - from + 1 }
         override fun toString() = "($from - $to, $lastProcessed)"
     }
 

src/main/java/org/radarbase/output/accounting/TopicPartitionOffsetRange.kt

@@ -30,22 +30,25 @@ data class TopicPartitionOffsetRange(
     val partition: Int = topicPartition.partition
 
     /** Full constructor.  */
-    constructor(topic: String, partition: Int, offsetFrom: Long, offsetTo: Long, lastModified: Instant = Instant.now()) : this(
+    constructor(topic: String, partition: Int, offsetFrom: Long, offsetTo: Long?, lastModified: Instant = Instant.now()) : this(
             TopicPartition(topic, partition),
             OffsetRangeSet.Range(offsetFrom, offsetTo, lastModified))
 
     override fun toString(): String = "$topic+$partition+${range.from}+${range.to} (${range.lastProcessed})"
 
     companion object {
+        private val filenameSplitRegex = "[+.]".toRegex()
+
         @Throws(NumberFormatException::class, IndexOutOfBoundsException::class)
         fun parseFilename(filename: String, lastModified: Instant): TopicPartitionOffsetRange {
-            val fileNameParts = filename.split("[+.]".toRegex()).dropLastWhile { it.isEmpty() }.toTypedArray()
+            val fileNameParts = filename.split(filenameSplitRegex)
+                    .dropLastWhile { it.isEmpty() || it == "avro" }
 
             return TopicPartitionOffsetRange(
                     fileNameParts[0],
-                    Integer.parseInt(fileNameParts[1]),
-                    java.lang.Long.parseLong(fileNameParts[2]),
-                    java.lang.Long.parseLong(fileNameParts[3]),
+                    fileNameParts[1].toInt(),
+                    fileNameParts[2].toLong(),
+                    fileNameParts.getOrNull(3)?.toLong(),
                     lastModified)
         }
     }

src/main/java/org/radarbase/output/config/RestructureConfig.kt

@@ -110,7 +110,12 @@ data class WorkerConfig(
          * Number of files to simultaneously keep in cache, including open writer. A higher size will
          * decrease overhead but increase memory usage and open file descriptors.
          */
-        val cacheSize: Int = CACHE_SIZE_DEFAULT) {
+        val cacheSize: Int = CACHE_SIZE_DEFAULT,
+        /**
+         * Number of offsets to simultaneously keep in cache. A higher size will
+         * decrease overhead but increase memory usage.
+         */
+        val cacheOffsetsSize: Long = 500_000) {
     init {
         check(cacheSize >= 1) { "Maximum files per topic must be strictly positive" }
         maxFilesPerTopic?.let { check(it >= 1) { "Maximum files per topic must be strictly positive" } }

src/main/java/org/radarbase/output/source/TopicFileList.kt

@@ -5,13 +5,15 @@ import java.nio.file.Path
 import java.time.Instant
 
 data class TopicFileList(val topic: String, val files: List<TopicFile>) {
-    val numberOfOffsets: Long = this.files.fold(0L) { sum, f -> sum + f.size }
+    val numberOfOffsets: Long? = this.files
+            .takeIf { fs -> fs.none { it.size == null } }
+            ?.fold(0L) { sum, f -> sum + f.size!! }
     val numberOfFiles: Int = this.files.size
 }
 
 data class TopicFile(val topic: String, val path: Path, val lastModified: Instant) {
     val range: TopicPartitionOffsetRange = TopicPartitionOffsetRange.parseFilename(path.fileName.toString(), lastModified)
-    val size: Long = range.range.size
+    val size: Long? = range.range.size
 }
 
 data class SimpleFileStatus(val path: Path, val isDirectory: Boolean, val lastModified: Instant)

src/main/java/org/radarbase/output/worker/RadarKafkaRestructure.kt

@@ -146,7 +146,12 @@ class RadarKafkaRestructure(
             .filter { f -> f.lastModified.isBefore(deleteThreshold) &&
                     // ensure that there is a file with a larger offset also
                     // processed, so the largest offset is never removed.
-                    seenFiles.contains(f.range.copy(range = f.range.range.copy(to = f.range.range.to + 1))) }
+                    if (f.range.range.to != null) {
+                        seenFiles.contains(f.range.copy(range = f.range.range.copy(to = f.range.range.to + 1)))
+                    } else {
+                        seenFiles.contains(f.range.topicPartition, f.range.range.from, f.range.range.lastProcessed)
+                    }
+            }
             .take(maxFilesPerTopic)
             .map { kafkaStorage.delete(it.path) }
             .count()

src/main/java/org/radarbase/output/worker/RestructureWorker.kt

@@ -35,47 +35,59 @@ internal class RestructureWorker(
     var processedRecordsCount: Long = 0
     private val reader = storage.reader()
     private val pathFactory: RecordPathFactory = fileStoreFactory.pathFactory
+    private val batchSize = fileStoreFactory.config.worker.cacheOffsetsSize
 
     private val cacheStore = fileStoreFactory.newFileCacheStore(accountant)
 
     fun processPaths(topicPaths: TopicFileList) {
         val numFiles = topicPaths.numberOfFiles
         val numOffsets = topicPaths.numberOfOffsets
+        val totalProgress = numOffsets ?: numFiles.toLong()
 
         val topic = topicPaths.topic
 
         val numberFormat = NumberFormat.getNumberInstance()
 
-        logger.info("Processing topic {}: converting {} files with {} records",
-                topic, numberFormat.format(numFiles), numberFormat.format(numOffsets))
+        if (numOffsets == null) {
+            logger.info("Processing topic {}: converting {} files",
+                    topic, numberFormat.format(numFiles))
+        } else {
+            logger.info("Processing topic {}: converting {} files with {} records",
+                    topic, numberFormat.format(numFiles), numberFormat.format(numOffsets))
+        }
 
         val seenOffsets = accountant.offsets
                 .withFactory { ReadOnlyFunctionalValue(it) }
 
-        val progressBar = ProgressBar(topic, numOffsets, 50, 5, TimeUnit.SECONDS)
+        val progressBar = ProgressBar(topic, totalProgress, 50, 5, TimeUnit.SECONDS)
         progressBar.update(0)
 
-        val batchSize = (BATCH_SIZE * ThreadLocalRandom.current().nextDouble(0.75, 1.25)).roundToLong()
-        var currentSize: Long = 0
+        val batchSize = generateBatchSize()
+        var currentSize = 0L
+        var currentFile = 0L
         try {
             for (file in topicPaths.files) {
                 if (closed.get()) {
                     break
                 }
-                try {
+                val processedSize = try {
                     this.processFile(file, progressBar, seenOffsets)
                 } catch (exc: JsonMappingException) {
                     logger.error("Cannot map values", exc)
+                    0L
                 }
 
-                processedFileCount++
-                progressBar.update(currentSize)
-
-                currentSize += file.size
+                currentFile += 1
+                currentSize += processedSize
                 if (currentSize >= batchSize) {
                     currentSize = 0
                     cacheStore.flush()
                 }
+
+                processedFileCount++
+                if (numOffsets == null) {
+                    progressBar.update(currentFile)
+                }
             }
         } catch (ex: IOException) {
             logger.error("Failed to process file", ex)
@@ -85,26 +97,28 @@ internal class RestructureWorker(
             logger.warn("Shutting down")
         }
 
-        progressBar.update(numOffsets)
+        progressBar.update(totalProgress)
     }
 
     @Throws(IOException::class)
     private fun processFile(file: TopicFile,
-                            progressBar: ProgressBar, seenOffsets: OffsetRangeSet) {
+                            progressBar: ProgressBar, seenOffsets: OffsetRangeSet): Long {
         logger.debug("Reading {}", file.path)
 
         val offset = file.range.range.from
 
-        reader.newInput(file).use { input ->
+        return reader.newInput(file).use { input ->
             // processing zero-length files may trigger a stall. See:
             // https://github.com/RADAR-base/Restructure-HDFS-topic/issues/3
             if (input.length() == 0L) {
                 logger.warn("File {} has zero length, skipping.", file.path)
-                return
+                return 0L
             }
             val transaction = Accountant.Transaction(file.range.topicPartition, offset, file.lastModified)
+            var recordsSeen = 0L
             extractRecords(input) { relativeOffset, record ->
                 transaction.offset = offset + relativeOffset
+                recordsSeen += 1L
                 val alreadyContains = time("accounting.check") {
                     seenOffsets.contains(file.range.topicPartition, transaction.offset, transaction.lastModified)
                 }
@@ -113,8 +127,11 @@ internal class RestructureWorker(
                     this.writeRecord(transaction, record)
                 }
                 processedRecordsCount++
-                progressBar.update(processedRecordsCount)
+                if (file.size != null) {
+                    progressBar.update(processedRecordsCount)
+                }
             }
+            recordsSeen
         }
     }
 
@@ -153,15 +170,17 @@ internal class RestructureWorker(
         }
     }
 
+    private fun generateBatchSize(): Long {
+        val modifier = ThreadLocalRandom.current().nextDouble(0.75, 1.25)
+        return (batchSize * modifier).roundToLong()
+    }
+
     override fun close() {
         reader.close()
         cacheStore.close()
     }
 
     companion object {
         private val logger = LoggerFactory.getLogger(RestructureWorker::class.java)
-
-        /** Number of offsets to process in a single task.  */
-        private const val BATCH_SIZE: Long = 500000
     }
 }