Merge pull request #78 from RADAR-base/release-1.1.1

nivemaham · web-flow · commit 98982cea9f69 · 2020-07-06T14:51:51.000+02:00
Release 1.1.1
diff --git a/README.md b/README.md
@@ -61,7 +61,7 @@ When upgrading to version 0.6.0 from version 0.5.x or earlier, please follow the
 
 This package is available as docker image [`radarbase/radar-output-restructure`](https://hub.docker.com/r/radarbase/radar-output-restructure). The entrypoint of the image is the current application. So in all the commands listed in usage, replace `radar-output-restructure` with for example:
 ```shell
-docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-output-restructure:1.1.0-hdfs -n hdfs-namenode -o /output /myTopic
+docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-output-restructure:1.1.1-hdfs -n hdfs-namenode -o /output /myTopic
 ```
 if your docker cluster is running in the `hadoop` network and your output directory should be `./output`.
 
@@ -171,7 +171,7 @@ This package requires at least Java JDK 8. Build the distribution with
 and install the package into `/usr/local` with for example
 ```shell
 sudo mkdir -p /usr/local
-sudo tar -xzf build/distributions/radar-output-restructure-1.1.0.tar.gz -C /usr/local --strip-components=1
+sudo tar -xzf build/distributions/radar-output-restructure-1.1.1.tar.gz -C /usr/local --strip-components=1
 ```
 
 Now the `radar-output-restructure` command should be available.
diff --git a/build.gradle b/build.gradle
@@ -9,7 +9,7 @@ plugins {
 }
 
 group 'org.radarbase'
-version '1.1.0'
+version '1.1.1'
 mainClassName = 'org.radarbase.output.Application'
 
 sourceCompatibility = '1.8'
diff --git a/restructure.yml b/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
diff --git a/src/integrationTest/java/org/radarbase/output/RestructureS3IntegrationTest.kt b/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()
diff --git a/src/main/java/org/radarbase/output/Application.kt b/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()
diff --git a/src/main/java/org/radarbase/output/cleaner/TimestampExtractionCheck.kt b/src/main/java/org/radarbase/output/cleaner/TimestampExtractionCheck.kt
@@ -61,7 +61,7 @@ class TimestampExtractionCheck(
                     TimestampFileCacheStore.FindResult.BAD_SCHEMA -> suffix += 1  // continue next suffix
                 }
             } catch (ex: IOException) {
-                logger.error("Failed to read target file for checking data integrity", ex)
+                logger.error("Failed to read target file {} for checking data integrity", path, ex)
                 return false
             }
         } while (true)
diff --git a/src/main/java/org/radarbase/output/cleaner/TimestampFileCache.kt b/src/main/java/org/radarbase/output/cleaner/TimestampFileCache.kt
@@ -53,7 +53,9 @@ class TimestampFileCache(
         if (header != null) {
             val recordHeader = converterFactory.headerFor(record)
             if (!recordHeader.contentEquals(header)) {
-                throw IllegalArgumentException("Header mismatch")
+                throw IllegalArgumentException(
+                        "Header mismatch: record header ${recordHeader.contentToString()}" +
+                                " does not match target header ${header.contentToString()}")
             }
         }
 
diff --git a/src/main/java/org/radarbase/output/cleaner/TimestampFileCacheStore.kt b/src/main/java/org/radarbase/output/cleaner/TimestampFileCacheStore.kt
@@ -19,6 +19,7 @@ package org.radarbase.output.cleaner
 import org.apache.avro.generic.GenericRecord
 import org.radarbase.output.FileStoreFactory
 import org.radarbase.output.util.Timer.time
+import org.slf4j.LoggerFactory
 import java.io.FileNotFoundException
 import java.io.IOException
 import java.nio.file.Path
@@ -60,13 +61,19 @@ class TimestampFileCacheStore(private val factory: FileStoreFactory) {
                     }
 
             time("cleaner.contains") {
-                if (fileCache.contains(record)) FindResult.FOUND else FindResult.NOT_FOUND
+                if (fileCache.contains(record)) FindResult.FOUND else {
+                    logger.warn("Target {} does not contain record {}", path, record)
+                    FindResult.NOT_FOUND
+                }
             }
         } catch (ex: FileNotFoundException) {
+            logger.warn("Target {} for {} has not been created yet.", path, record)
             FindResult.FILE_NOT_FOUND
         } catch (ex: IllegalArgumentException) {
+            logger.warn("Schema of {} does not match schema of record {}: {}", path, record, ex.message)
             FindResult.BAD_SCHEMA
         } catch (ex: IndexOutOfBoundsException) {
+            logger.warn("Schema of {} does not match schema of record {} (wrong number of columns)", path, record)
             FindResult.BAD_SCHEMA
         }
     }
@@ -95,4 +102,8 @@ class TimestampFileCacheStore(private val factory: FileStoreFactory) {
         NOT_FOUND,
         FOUND
     }
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(TimestampFileCacheStore::class.java)
+    }
 }
diff --git a/src/main/java/org/radarbase/output/config/RestructureConfig.kt b/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" } }
diff --git a/src/main/java/org/radarbase/output/util/TimeUtil.kt b/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())
 }
diff --git a/src/main/java/org/radarbase/output/worker/RadarKafkaRestructure.kt b/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())
 
diff --git a/src/main/java/org/radarbase/output/worker/RestructureWorker.kt b/src/main/java/org/radarbase/output/worker/RestructureWorker.kt
@@ -3,6 +3,7 @@ package org.radarbase.output.worker
 import com.fasterxml.jackson.databind.JsonMappingException
 import org.apache.avro.file.DataFileReader
 import org.apache.avro.file.SeekableInput
+import org.apache.avro.generic.GenericData
 import org.apache.avro.generic.GenericDatumReader
 import org.apache.avro.generic.GenericRecord
 import org.radarbase.output.FileStoreFactory
@@ -176,8 +177,11 @@ internal class RestructureWorker(
 
         fun <T> extractRecords(input: SeekableInput, processing: (Sequence<GenericRecord>) -> T): T {
             var tmpRecord: GenericRecord? = null
+            val genericData = GenericData().apply {
+                isFastReaderEnabled = true
+            }
 
-            return DataFileReader(input, GenericDatumReader<GenericRecord>()).use { reader ->
+            return DataFileReader(input, GenericDatumReader<GenericRecord>(null, null, genericData)).use { reader ->
                 processing(generateSequence {
                     time("read") {
                         if (reader.hasNext()) reader.next(tmpRecord) else null
diff --git a/src/test/java/org/radarbase/output/cleaner/TimestampFileCacheTest.kt b/src/test/java/org/radarbase/output/cleaner/TimestampFileCacheTest.kt
@@ -0,0 +1,108 @@
+package org.radarbase.output.cleaner
+
+import com.nhaarman.mockitokotlin2.doReturn
+import com.nhaarman.mockitokotlin2.mock
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericData
+import org.apache.avro.generic.GenericRecord
+import org.apache.avro.generic.GenericRecordBuilder
+import org.hamcrest.MatcherAssert.assertThat
+import org.hamcrest.Matchers.`is`
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import org.junit.jupiter.api.io.TempDir
+import org.radarbase.output.FileStoreFactory
+import org.radarbase.output.compression.IdentityCompression
+import org.radarbase.output.config.LocalConfig
+import org.radarbase.output.format.CsvAvroConverterFactory
+import org.radarbase.output.target.LocalTargetStorage
+import java.io.ByteArrayInputStream
+import java.io.FileNotFoundException
+import java.io.InputStreamReader
+import java.nio.file.Files
+import java.nio.file.Path
+
+internal class TimestampFileCacheTest {
+    private lateinit var record: GenericData.Record
+    private var now: Double = 0.0
+    private lateinit var schema: Schema
+    private lateinit var factory: FileStoreFactory
+    private lateinit var csvConverter: CsvAvroConverterFactory
+
+    @BeforeEach
+    fun setUp() {
+        csvConverter = CsvAvroConverterFactory()
+        factory = mock {
+            on { recordConverter } doReturn csvConverter
+            on { targetStorage } doReturn LocalTargetStorage(LocalConfig())
+            on { compression } doReturn IdentityCompression()
+        }
+        schema = Schema.Parser().parse(javaClass.getResourceAsStream("android_phone_light.avsc"))
+        now = System.currentTimeMillis() / 1000.0
+        record = GenericRecordBuilder(schema)
+                .set("key", GenericRecordBuilder(schema.getField("key")!!.schema())
+                        .set("projectId", "p")
+                        .set("userId", "u")
+                        .set("sourceId", "s")
+                        .build())
+                .set("value", GenericRecordBuilder(schema.getField("value")!!.schema())
+                        .set("time", now)
+                        .set("timeReceived", now + 1.0)
+                        .set("light", 1.0f)
+                        .build())
+                .build()
+    }
+
+    @Test
+    fun testFileCacheFound(@TempDir path: Path) {
+        val targetPath = path.resolve("test.avro")
+        writeRecord(targetPath, record)
+        val timestampFileCache = TimestampFileCache(factory, targetPath)
+        assertThat(timestampFileCache.contains(record), `is`(true))
+    }
+
+    private fun writeRecord(path: Path, record: GenericRecord) {
+        Files.newBufferedWriter(path).use { wr ->
+            ByteArrayInputStream(ByteArray(0)).use { emptyInput ->
+                InputStreamReader(emptyInput).use { emptyReader ->
+                    csvConverter.converterFor(wr, record, true, emptyReader).use { converter ->
+                        converter.writeRecord(record)
+                    }
+                }
+            }
+        }
+    }
+
+    @Test
+    fun testFileCacheNotFound(@TempDir path: Path) {
+        val targetPath = path.resolve("test.avro")
+        assertThrows<FileNotFoundException> { TimestampFileCache(factory, targetPath) }
+    }
+
+    @Test
+    fun testHeaderMismatch(@TempDir path: Path) {
+        val targetPath = path.resolve("test.avro")
+        Files.newBufferedWriter(targetPath).use { writer ->
+            writer.write("key.projectId,key.userId,key.sourceId,value.time,value.timeReceived,value.luminance")
+        }
+        val cache = TimestampFileCache(factory, targetPath)
+        assertThrows<IllegalArgumentException> { cache.contains(record) }
+    }
+
+    @Test
+    fun testNotFound(@TempDir path: Path) {
+        val targetPath = path.resolve("test.avro")
+
+        val otherRecord = GenericRecordBuilder(record)
+                .set("value", GenericRecordBuilder(record.get("value") as GenericData.Record)
+                        .set("time", now + 1.0)
+                        .set("timeReceived", now + 2.0)
+                        .build())
+                .build()
+
+        writeRecord(targetPath, otherRecord)
+        val cache = TimestampFileCache(factory, targetPath)
+        assertThat(cache.contains(record), `is`(false))
+    }
+}
diff --git a/src/test/resources/org/radarbase/output/cleaner/android_phone_light.avsc b/src/test/resources/org/radarbase/output/cleaner/android_phone_light.avsc

Original file line number	Diff line number	Diff line change
`@@ -9,7 +9,7 @@ plugins {`
`9`	`9`	`}`
`10`	`10`
`11`	`11`	`group 'org.radarbase'`
`12`		`-version '1.1.0'`
	`12`	`+version '1.1.1'`
`13`	`13`	`mainClassName = 'org.radarbase.output.Application'`
`14`	`14`
`15`	`15`	`sourceCompatibility = '1.8'`
Original file line number	Diff line number	Diff line change
`@@ -61,7 +61,7 @@ class TimestampExtractionCheck(`
`61`	`61`	`TimestampFileCacheStore.FindResult.BAD_SCHEMA -> suffix += 1 // continue next suffix`
`62`	`62`	`}`
`63`	`63`	`} catch (ex: IOException) {`
`64`		`- logger.error("Failed to read target file for checking data integrity", ex)`
	`64`	`+ logger.error("Failed to read target file {} for checking data integrity", path, ex)`
`65`	`65`	`return false`
`66`	`66`	`}`
`67`	`67`	`} while (true)`
Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,9 @@ class TimestampFileCache(`
`53`	`53`	`if (header != null) {`
`54`	`54`	`val recordHeader = converterFactory.headerFor(record)`
`55`	`55`	`if (!recordHeader.contentEquals(header)) {`
`56`		`- throw IllegalArgumentException("Header mismatch")`
	`56`	`+ throw IllegalArgumentException(`
	`57`	`+ "Header mismatch: record header ${recordHeader.contentToString()}" +`
	`58`	`+ " does not match target header ${header.contentToString()}")`
`57`	`59`	`}`
`58`	`60`	`}`
`59`	`61`