Merge pull request #77 from RADAR-base/skipRecentFiles

Only process files if they are a minimum age.

Author: blootsvoets

restructure.yml

@@ -79,6 +79,10 @@ worker:
   numThreads: 2
   # Maximum number of files to process in any given topic.
   maxFilesPerTopic: null
+  # Minimum time in seconds since a source file was last modified before it is
+  # considered for processing. This avoids synchronization issues when a file has just been
+  # written.
+  minimumFileAge: 60
 
 cleaner:
   # Enable cleaning up old source files

src/integrationTest/java/org/radarbase/output/RestructureS3IntegrationTest.kt

@@ -4,10 +4,7 @@ import io.minio.PutObjectOptions
 import io.minio.PutObjectOptions.MAX_PART_SIZE
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
-import org.radarbase.output.config.PathConfig
-import org.radarbase.output.config.ResourceConfig
-import org.radarbase.output.config.RestructureConfig
-import org.radarbase.output.config.S3Config
+import org.radarbase.output.config.*
 import org.radarbase.output.util.Timer
 import java.nio.charset.StandardCharsets.UTF_8
 import java.nio.file.Paths
@@ -29,7 +26,8 @@ class RestructureS3IntegrationTest {
         val config = RestructureConfig(
                 source = ResourceConfig("s3", s3 = sourceConfig),
                 target = ResourceConfig("s3", s3 = targetConfig),
-                paths = PathConfig(inputs = listOf(Paths.get("in")))
+                paths = PathConfig(inputs = listOf(Paths.get("in"))),
+                worker = WorkerConfig(minimumFileAge = 0L)
         )
         val application = Application(config)
         val sourceClient = sourceConfig.createS3Client()

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

@@ -32,6 +32,7 @@ import org.radarbase.output.util.ProgressBar.Companion.format
 import org.radarbase.output.util.Timer
 import org.radarbase.output.worker.FileCacheStore
 import org.radarbase.output.cleaner.SourceDataCleaner
+import org.radarbase.output.util.TimeUtil.durationSince
 import org.radarbase.output.worker.RadarKafkaRestructure
 import org.slf4j.LoggerFactory
 import redis.clients.jedis.JedisPool
@@ -187,8 +188,6 @@ class Application(
         private val logger = LoggerFactory.getLogger(Application::class.java)
         const val CACHE_SIZE_DEFAULT = 100
 
-        private fun Temporal.durationSince() = Duration.between(this, Instant.now())
-
         private fun parseArgs(args: Array<String>): CommandLineArgs {
             val commandLineArgs = CommandLineArgs()
             JCommander.newBuilder()

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

@@ -143,7 +143,14 @@ data class WorkerConfig(
          * Number of offsets to simultaneously keep in cache. A higher size will
          * decrease overhead but increase memory usage.
          */
-        val cacheOffsetsSize: Long = 500_000) {
+        val cacheOffsetsSize: Long = 500_000,
+        /**
+         * Minimum time since the file was last modified in seconds. Avoids
+         * synchronization issues that may occur in a source file that is being
+         * appended to.
+         */
+        val minimumFileAge: Long = 60
+) {
     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/util/TimeUtil.kt

@@ -4,11 +4,9 @@ import com.fasterxml.jackson.databind.JsonNode
 import org.apache.avro.Schema
 import org.apache.avro.generic.GenericRecord
 import java.math.RoundingMode
-import java.time.Instant
-import java.time.LocalDate
-import java.time.LocalDateTime
-import java.time.ZoneOffset
+import java.time.*
 import java.time.format.DateTimeParseException
+import java.time.temporal.Temporal
 
 object TimeUtil {
     private val NANO_MULTIPLIER = 1_000_000_000.toBigDecimal()
@@ -143,4 +141,7 @@ object TimeUtil {
     fun Instant.toDouble() = (epochSecond.toBigDecimal()
             + (nano.toBigDecimal().divide(NANO_MULTIPLIER, 9, RoundingMode.HALF_UP))
             ).toDouble()
+
+
+    fun Temporal.durationSince(): Duration = Duration.between(this, Instant.now())
 }

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

@@ -20,11 +20,13 @@ import org.radarbase.output.FileStoreFactory
 import org.radarbase.output.accounting.Accountant
 import org.radarbase.output.accounting.OffsetRangeSet
 import org.radarbase.output.source.TopicFileList
+import org.radarbase.output.util.TimeUtil.durationSince
 import org.slf4j.LoggerFactory
 import java.io.Closeable
 import java.io.IOException
 import java.nio.file.Path
 import java.nio.file.Paths
+import java.time.Duration
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.LongAdder
 
@@ -39,17 +41,28 @@ import java.util.concurrent.atomic.LongAdder
 class RadarKafkaRestructure(
         private val fileStoreFactory: FileStoreFactory
 ): Closeable {
-    private val maxFilesPerTopic: Int = fileStoreFactory.config.worker.maxFilesPerTopic ?: Int.MAX_VALUE
     private val sourceStorage = fileStoreFactory.sourceStorage
 
     private val lockManager = fileStoreFactory.remoteLockManager
-    private val excludeTopics: Set<String> = fileStoreFactory.config.topics
-            .mapNotNullTo(HashSet()) { (topic, conf) ->
-                topic.takeIf { conf.exclude }
-            }
 
     private val isClosed = AtomicBoolean(false)
 
+    private val excludeTopics: Set<String>
+    private val maxFilesPerTopic: Int
+    private val minimumFileAge: Duration
+
+    init {
+        val config = fileStoreFactory.config
+        excludeTopics = config.topics
+                .mapNotNullTo(HashSet()) { (topic, conf) ->
+                    topic.takeIf { conf.exclude }
+                }
+
+        val workerConfig = config.worker
+        maxFilesPerTopic = workerConfig.maxFilesPerTopic ?: Int.MAX_VALUE
+        minimumFileAge = Duration.ofSeconds(workerConfig.minimumFileAge.coerceAtLeast(0L))
+    }
+
     val processedFileCount = LongAdder()
     val processedRecordsCount = LongAdder()
 
@@ -103,7 +116,8 @@ class RadarKafkaRestructure(
         return RestructureWorker(sourceStorage, accountant, fileStoreFactory, isClosed).use { worker ->
             try {
                 val topicPaths = TopicFileList(topic, sourceStorage.walker.walkRecords(topic, topicPath)
-                        .filter { f -> !seenFiles.contains(f.range) }
+                        .filter { f -> !seenFiles.contains(f.range)
+                                && f.lastModified.durationSince() >= minimumFileAge }
                         .take(maxFilesPerTopic)
                         .toList())