Merge branch 'dev' of github.com:RADAR-base/Restructure-HDFS-topic into dev

Author: nivemaham

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/worker/RestructureWorker.kt

@@ -35,6 +35,7 @@ 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)
 
@@ -61,9 +62,9 @@ internal class RestructureWorker(
         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 = 0L
-        var currentFile: Long = 0L
+        val batchSize = generateBatchSize()
+        var currentSize = 0L
+        var currentFile = 0L
         try {
             for (file in topicPaths.files) {
                 if (closed.get()) {
@@ -169,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
     }
 }